Compare commits

..

4 Commits

5 changed files with 33 additions and 25 deletions

View File

@ -1,9 +1,11 @@
package main package main
import "os" import "os"
import "io"
import "fmt" import "fmt"
import "time" import "time"
import "bufio" import "bufio"
import "errors"
import "context" import "context"
import "crypto/tls" import "crypto/tls"
import "git.tebibyte.media/sashakoshka/hopp" import "git.tebibyte.media/sashakoshka/hopp"
@ -22,6 +24,7 @@ func main() {
} }
trans, err := join(address, room, nickname) trans, err := join(address, room, nickname)
handleErr(1, err) handleErr(1, err)
fmt.Fprintf(os.Stdout, "(i) connected to %s/%s\n", address, room)
go func() { go func() {
reader := bufio.NewReader(os.Stdin) reader := bufio.NewReader(os.Stdin)
for { for {
@ -32,7 +35,12 @@ func main() {
}() }()
for { for {
message, _, err := chat.Receive(trans) message, _, err := chat.Receive(trans)
if err != nil {
if !errors.Is(err, io.EOF) {
handleErr(1, err) handleErr(1, err)
}
break
}
switch message := message.(type) { switch message := message.(type) {
case *chat.MessageChat: case *chat.MessageChat:
fmt.Fprintf(os.Stdout, "%s: %s\n", message.Nickname, message.Content) fmt.Fprintf(os.Stdout, "%s: %s\n", message.Nickname, message.Content)
@ -42,6 +50,7 @@ func main() {
fmt.Fprintf(os.Stdout, "(i) %s left the room\n", message.Nickname) fmt.Fprintf(os.Stdout, "(i) %s left the room\n", message.Nickname)
} }
} }
fmt.Fprintf(os.Stdout, "(i) disconnected\n")
} }
func join(address string, room string, nickname string) (hopp.Trans, error) { func join(address string, room string, nickname string) (hopp.Trans, error) {

View File

@ -1,6 +1,7 @@
package main package main
import "os" import "os"
import "io"
import "fmt" import "fmt"
import "log" import "log"
import "errors" import "errors"
@ -48,7 +49,7 @@ func host(address string, certPath, keyPath string) error {
type client struct { type client struct {
conn hopp.Conn conn hopp.Conn
nickname hopp.Option[string] nickname string
rooms usync.RWMonitor[map[string] hopp.Trans] rooms usync.RWMonitor[map[string] hopp.Trans]
} }
@ -58,13 +59,13 @@ func (this *client) run() {
defer this.conn.Close() defer this.conn.Close()
for { for {
log.Println("accepting transaction")
trans, err := this.conn.AcceptTrans() trans, err := this.conn.AcceptTrans()
if err != nil { if err != nil {
if !errors.Is(err, io.EOF) {
log.Printf("XXX %v failed: %v", this.conn.RemoteAddr(), err) log.Printf("XXX %v failed: %v", this.conn.RemoteAddr(), err)
}
return return
} }
log.Println("accepted transaction")
go this.runTrans(trans) go this.runTrans(trans)
} }
} }
@ -111,7 +112,7 @@ func (this *client) transTalk(trans hopp.Trans, initial *chat.MessageJoin) error
} }
func (this *client) handleMessageChat(trans hopp.Trans, room string, message *chat.MessageChat) error { func (this *client) handleMessageChat(trans hopp.Trans, room string, message *chat.MessageChat) error {
log.Println("(). %s #%s: %s", this.nickname, room, message.Content) log.Printf("(). %s #%s: %s", this.nickname, room, message.Content)
clients, done := clients.RBorrow() clients, done := clients.RBorrow()
defer done() defer done()
for client := range clients { for client := range clients {

View File

@ -1237,7 +1237,9 @@ func (this *Generator) generateReceive() (n int, err error) {
} }
nn, err = this.iprint("}\n") nn, err = this.iprint("}\n")
n += nn; if err != nil { return n, err } n += nn; if err != nil { return n, err }
nn, err = this.iprint("return nil, n, fmt.Errorf(\"%w: M%04X\", hopp.ErrUnknownMethod, method)\n") // fuck off go vet
str := `return nil, n, fmt.Errorf("%w: M%04X", hopp.ErrUnknownMethod, method)`
nn, err = this.iprintln(str)
n += nn; if err != nil { return n, err } n += nn; if err != nil { return n, err }
this.pop() this.pop()
nn, err = this.iprint("}\n") nn, err = this.iprint("}\n")

View File

@ -6,6 +6,7 @@ import "fmt"
import "net" import "net"
import "sync" import "sync"
import "time" import "time"
import "context"
import "sync/atomic" import "sync/atomic"
import "git.tebibyte.media/sashakoshka/go-util/sync" import "git.tebibyte.media/sashakoshka/go-util/sync"
@ -39,20 +40,23 @@ type a struct {
sendLock sync.Mutex sendLock sync.Mutex
transMap map[int64] *transA transMap map[int64] *transA
transChan chan *transA transChan chan *transA
done chan struct { } ctx context.Context
done func()
err error err error
} }
// AdaptA returns a connection implementing METADAPT-A over a singular stream- // AdaptA returns a connection implementing METADAPT-A over a singular stream-
// oriented transport such as TCP or UNIX domain stream sockets. // oriented transport such as TCP or UNIX domain stream sockets.
func AdaptA(underlying net.Conn, party Party) Conn { func AdaptA(underlying net.Conn, party Party) Conn {
ctx, done := context.WithCancel(context.Background())
conn := &a { conn := &a {
sizeLimit: defaultSizeLimit, sizeLimit: defaultSizeLimit,
underlying: underlying, underlying: underlying,
party: party, party: party,
transMap: make(map[int64] *transA), transMap: make(map[int64] *transA),
transChan: make(chan *transA), transChan: make(chan *transA),
done: make(chan struct { }), ctx: ctx,
done: done,
} }
if party == ClientSide { if party == ClientSide {
conn.transID = 1 conn.transID = 1
@ -60,11 +64,15 @@ func AdaptA(underlying net.Conn, party Party) Conn {
conn.transID = -1 conn.transID = -1
} }
go conn.receive() go conn.receive()
go func() {
<- ctx.Done()
underlying.Close()
}()
return conn return conn
} }
func (this *a) Close() error { func (this *a) Close() error {
close(this.done) this.done()
return nil return nil
} }
@ -105,7 +113,7 @@ func (this *a) AcceptTrans() (Trans, error) {
return nil, eof return nil, eof
} }
return trans, nil return trans, nil
case <- this.done: case <- this.ctx.Done():
return nil, eof return nil, eof
} }
} }
@ -490,6 +498,7 @@ func decodeMessageA(
headerBuffer := [18]byte { } headerBuffer := [18]byte { }
_, err = io.ReadFull(reader, headerBuffer[:]) _, err = io.ReadFull(reader, headerBuffer[:])
if err != nil { return 0, 0, false, nil, err } if err != nil { return 0, 0, false, nil, err }
transID, err = decodeI64[int64](headerBuffer[:8]) transID, err = decodeI64[int64](headerBuffer[:8])
if err != nil { return 0, 0, false, nil, err } if err != nil { return 0, 0, false, nil, err }
method, err = decodeI16[uint16](headerBuffer[8:10]) method, err = decodeI16[uint16](headerBuffer[8:10])

View File

@ -52,6 +52,7 @@ func TestConnA(test *testing.T) {
test.Error("CLIENT payload:", gotPayload) test.Error("CLIENT payload:", gotPayload)
test.Fatal("CLIENT ok byeeeeeeeeeeeee") test.Fatal("CLIENT ok byeeeeeeeeeeeee")
} }
test.Log("CLIENT transaction has closed")
} }
serverFunc := func(a Conn) { serverFunc := func(a Conn) {
@ -70,20 +71,6 @@ func TestConnA(test *testing.T) {
} }
func TestTransOpenCloseA(test *testing.T) { func TestTransOpenCloseA(test *testing.T) {
// currently:
//
// | data sent | data recvd | close sent | close recvd
// 10 | X | X | X | server hangs
// 20 | X | X | X | client hangs
// 30 | X | | X |
//
// when a close message is recvd, it tries to push to the trans and
// hangs on trans.incoming.Send, which hangs on sending the value to the
// underlying channel. why is this?
//
// check if we are really getting values from the channel when pulling
// from the trans channel when we are expecting a close.
clientFunc := func(conn Conn) { clientFunc := func(conn Conn) {
// 10 // 10
trans, err := conn.OpenTrans() trans, err := conn.OpenTrans()