From 688668f6d34d4d962247bd5df60c3fca61185927 Mon Sep 17 00:00:00 2001 From: Kyren223 Date: Thu, 24 Oct 2024 16:39:14 +0300 Subject: feat: worked on client --- internal/server/server.go | 74 +++++++++++++++++++++++++++++++++-------------- 1 file changed, 53 insertions(+), 21 deletions(-) (limited to 'internal/server') diff --git a/internal/server/server.go b/internal/server/server.go index 4ac8969..ff28ad7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -168,6 +168,9 @@ func handleConnection(ctx context.Context, conn net.Conn, server *server) { if !ok { break } + if packet.Type().IsPush() { + log.Println(addr, "streaming packet:", packet) + } if _, err := packet.Into(conn); err != nil { log.Println(addr, err) break @@ -192,7 +195,8 @@ func handleConnection(ctx context.Context, conn net.Conn, server *server) { buffer := make([]byte, 512) for { - conn.SetReadDeadline(time.Now().Add(time.Second)) + err := conn.SetReadDeadline(time.Now().Add(time.Second)) + assert.NoError(err, "setting read deadline should not error") n, err := conn.Read(buffer) deadlineExceeded := errors.Is(err, os.ErrDeadlineExceeded) if err != nil && !deadlineExceeded { @@ -220,13 +224,19 @@ func handleConnection(ctx context.Context, conn net.Conn, server *server) { } func handleAuth(conn net.Conn, nonce []byte) (ed25519.PublicKey, error) { - conn.SetDeadline(time.Now().Add(time.Second * 5)) + err := conn.SetDeadline(time.Now().Add(time.Second * 5)) + assert.NoError(err, "setting read deadline should not error") + + defer func() { + err := conn.SetDeadline(time.Time{}) + assert.NoError(err, "unsetting read deadline should not error") + }() challengePacket := make([]byte, len(nonce)+1) challengePacket[0] = packet.VERSION copy(challengePacket[1:], nonce) - _, err := conn.Write(challengePacket) + _, err = conn.Write(challengePacket) if err != nil { return nil, fmt.Errorf("error writing challenge: %w", err) } @@ -252,7 +262,6 @@ func handleAuth(conn net.Conn, nonce []byte) (ed25519.PublicKey, error) { return nil, errors.New("signature verification failed") } - conn.SetDeadline(time.Time{}) return pubKey, nil } @@ -310,20 +319,31 @@ func processPacket(ctx context.Context, pkt packet.Packet) packet.Packet { session, ok := FromContext(ctx) assert.Assert(ok, "context in process packet should always have a session") - var response packet.TypedMessage - switch pkt.Type() { - case packet.PacketSendMessage: - var request packet.SendMessage - if err := pkt.DecodePayload(&request); err != nil { - log.Println("decode error:", err) - response = &packet.ErrorMessage{Error: "malformed payload"} - break - } + var response packet.Payload + + request, err := pkt.DecodedPayload() + if err != nil { + response = &packet.ErrorMessage{Error: "malformed payload"} + } else { + response = processRequest(ctx, request) + } + + assert.NotNil(response, "response must always be assigned to") + log.Println(session.Addr, "sending ", response.Type(), "response:", response) + return packet.NewPacket(packet.NewMsgPackEncoder(response)) +} + +func processRequest(ctx context.Context, request packet.Payload) packet.Payload { + session, ok := FromContext(ctx) + assert.Assert(ok, "context in process packet should always have a session") + + log.Println(session.Addr, "processing", request.Type(), "request:", request) + switch request := request.(type) { + case *packet.SendMessage: content := strings.TrimSpace(request.Content) if content == "" { - response = &packet.ErrorMessage{Error: "message content must not be blank"} - break + return &packet.ErrorMessage{Error: "message content must not be blank"} } node := session.Server.Node @@ -334,15 +354,27 @@ func processPacket(ctx context.Context, pkt packet.Packet) packet.Packet { NetworkId: node.Generate(), // TODO: replace with actual ID Contents: content, } - messages = append(messages, message) + sendMessage(ctx, message) + + return packet.NewOkMessage() - response = packet.NewOkMessage() default: - response = &packet.ErrorMessage{Error: "use of unsupported packet type"} + return &packet.ErrorMessage{Error: "use of unsupported packet type"} } - - assert.NotNil(response, "response must always be assigned to") - return packet.NewPacket(packet.NewMsgPackEncoder(response)) } var messages []data.Message + +func sendMessage(ctx context.Context, msg data.Message) { + session, ok := FromContext(ctx) + assert.Assert(ok, "context in process packet should always have a session") + + messages = append(messages, msg) + + payload := &packet.Messages{ + Messages: messages, + } + + pkt := packet.NewPacket(packet.NewMsgPackEncoder(payload)) + session.WriteQueue <- pkt +} -- cgit v1.3.1