Reorganizing where quic and rest of transports will be

benchmark/quic.go

@@ -7,14 +7,15 @@ import (
 	"fmt"
 	"github.com/quic-go/quic-go"
 	"github.com/unpackdev/fdb"
+	transport_quic "github.com/unpackdev/fdb/transports/quic"
 	"github.com/unpackdev/fdb/types"
 	"io"
 )
 
 // QuicSuite represents the QUIC-specific benchmark suite.
 type QuicSuite struct {
 	fdbInstance *fdb.FDB
-	quicServer  *fdb.QuicServer
+	quicServer  *transport_quic.Server
 }
 
 // NewQuicSuite creates a new QuicSuite for benchmarking.
@@ -31,7 +32,7 @@ func (qs *QuicSuite) Start() error {
 		return fmt.Errorf("failed to retrieve QUIC transport: %w", err)
 	}
 
-	quicServer, ok := quicTransport.(*fdb.QuicServer)
+	quicServer, ok := quicTransport.(*transport_quic.Server)
 	if !ok {
 		return fmt.Errorf("failed to cast transport to QuicServer")
 	}
@@ -41,11 +42,11 @@ func (qs *QuicSuite) Start() error {
 		return fmt.Errorf("failed to retrieve benchmark database: %w", err)
 	}
 
-	wHandler := fdb.NewQuicWriteHandler(db)
-	quicServer.RegisterHandler(fdb.WriteHandlerType, wHandler.HandleMessage)
+	wHandler := transport_quic.NewQuicWriteHandler(db)
+	quicServer.RegisterHandler(types.WriteHandlerType, wHandler.HandleMessage)
 
-	rHandler := fdb.NewQuicReadHandler(db)
-	quicServer.RegisterHandler(fdb.ReadHandlerType, rHandler.HandleMessage)
+	rHandler := transport_quic.NewQuicReadHandler(db)
+	quicServer.RegisterHandler(types.ReadHandlerType, rHandler.HandleMessage)
 
 	if err := quicServer.Start(); err != nil {
 		return fmt.Errorf("failed to start QUIC server: %w", err)

fdb.go

@@ -5,6 +5,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/unpackdev/fdb/config"
 	"github.com/unpackdev/fdb/db"
+	transport_quic "github.com/unpackdev/fdb/transports/quic"
 	"github.com/unpackdev/fdb/types"
 )
 
@@ -38,7 +39,7 @@ func New(ctx context.Context, cnf config.Config) (*FDB, error) {
 	for _, transport := range cnf.Transports {
 		switch t := transport.Config.(type) {
 		case config.QuicTransport:
-			quicServer, err := NewQuicServer(ctx, t)
+			quicServer, err := transport_quic.NewServer(ctx, t)
 			if err != nil {
 				return nil, errors.Wrap(err, "failed to create QUIC server")
 			}

messages/message.go

@@ -0,0 +1,62 @@
+package messages
+
+import (
+	"encoding/binary"
+	"fmt"
+	"github.com/unpackdev/fdb/types"
+)
+
+// Message struct represents a UDP message
+type Message struct {
+	Handler types.HandlerType // The handler type (1 byte)
+	Key     [32]byte          // Fixed 32-byte key (e.g., Ethereum hash)
+	Data    []byte            // The remaining data after the key
+}
+
+// Encode encodes the Message struct into a byte slice
+func (m *Message) Encode() ([]byte, error) {
+	// Add 4 bytes for the length of the data
+	msgLen := 1 + 32 + 4 + len(m.Data)
+	buf := make([]byte, msgLen)
+
+	// Set handler type
+	buf[0] = byte(m.Handler)
+
+	// Copy the 32-byte key
+	copy(buf[1:33], m.Key[:])
+
+	// Set the length of the data (4 bytes)
+	binary.BigEndian.PutUint32(buf[33:37], uint32(len(m.Data)))
+
+	// Copy the data
+	copy(buf[37:], m.Data)
+
+	return buf, nil
+}
+
+// Decode decodes a byte slice into a Message struct
+func Decode(data []byte) (*Message, error) {
+	if len(data) < 37 { // 1 byte for handler + 32 bytes for key + 4 bytes for data length
+		return nil, fmt.Errorf("data too short, must be at least 37 bytes")
+	}
+
+	msg := &Message{
+		Handler: types.HandlerType(data[0]),
+	}
+
+	// Copy the 32-byte key
+	copy(msg.Key[:], data[1:33])
+
+	// Read the 4-byte data length
+	dataLen := binary.BigEndian.Uint32(data[33:37])
+
+	// Ensure the length of the remaining data matches the declared length
+	if len(data[37:]) < int(dataLen) {
+		return nil, fmt.Errorf("data length mismatch, expected %d bytes but got %d bytes", dataLen, len(data[37:]))
+	}
+
+	// Copy the data
+	msg.Data = data[37 : 37+dataLen]
+
+	return msg, nil
+}

quic_handler_read.go → transports/quic/handler_read.go

@@ -1,9 +1,10 @@
-package fdb
+package transport_quic
 
 import (
 	"encoding/binary"
 	"github.com/quic-go/quic-go"
 	"github.com/unpackdev/fdb/db"
+	"github.com/unpackdev/fdb/messages"
 	"log"
 )
 
@@ -20,7 +21,7 @@ func NewQuicReadHandler(db db.Provider) *QuicReadHandler {
 }
 
 // HandleMessage processes the incoming message using the QuicReadHandler
-func (rh *QuicReadHandler) HandleMessage(conn quic.Connection, stream quic.Stream, message *Message) {
+func (rh *QuicReadHandler) HandleMessage(conn quic.Connection, stream quic.Stream, message *messages.Message) {
 	//log.Printf("Processing read request: Handler=%d, Key=%x", message.Handler, message.Key)
 
 	// Query the database using the key from the Message struct

quic_handler_read_test.go → transports/quic/handler_read_test.go

@@ -1,4 +1,4 @@
-package fdb
+package transport_quic
 
 import (
 	"context"

quic_handler_write.go → transports/quic/handler_write.go

@@ -1,8 +1,9 @@
-package fdb
+package transport_quic
 
 import (
 	"github.com/quic-go/quic-go"
 	"github.com/unpackdev/fdb/db"
+	"github.com/unpackdev/fdb/messages"
 	"log"
 )
 
@@ -19,7 +20,7 @@ func NewQuicWriteHandler(db db.Provider) *QuicWriteHandler {
 }
 
 // HandleMessage processes the incoming message using the QuicWriteHandler
-func (wh *QuicWriteHandler) HandleMessage(conn quic.Connection, stream quic.Stream, message *Message) {
+func (wh *QuicWriteHandler) HandleMessage(conn quic.Connection, stream quic.Stream, message *messages.Message) {
 	// Log the message for debugging purposes
 	//log.Printf("Processing write request: Handler=%d, Key=%x, Data=%s", message.Handler, message.Key, string(message.Data))
 

quic_handler_write_test.go → transports/quic/handler_write_test.go

@@ -1,4 +1,4 @@
-package fdb
+package transport_quic
 
 import (
 	"context"

quic_server.go → transports/quic/server.go

@@ -1,4 +1,4 @@
-package fdb
+package transport_quic
 
 import (
 	"context"
@@ -7,19 +7,21 @@ import (
 	"github.com/pkg/errors"
 	"github.com/quic-go/quic-go"
 	"github.com/unpackdev/fdb/config"
+	"github.com/unpackdev/fdb/messages"
+	"github.com/unpackdev/fdb/types"
 	"io"
 	"log"
 	"strings"
 	"sync"
 )
 
 // QuicHandler function type for QUIC
-type QuicHandler func(sess quic.Connection, stream quic.Stream, message *Message)
+type QuicHandler func(sess quic.Connection, stream quic.Stream, message *messages.Message)
 
-// QuicServer struct represents the QUIC server
-type QuicServer struct {
+// Server struct represents the QUIC server
+type Server struct {
 	ctx             context.Context
-	handlerRegistry map[HandlerType]QuicHandler
+	handlerRegistry map[types.HandlerType]QuicHandler
 	cnf             config.QuicTransport
 	tlsConfig       *tls.Config
 	stopChan        chan struct{}
@@ -28,16 +30,16 @@ type QuicServer struct {
 	listener        *quic.Listener
 }
 
-// NewQuicServer creates a new QuicServer instance
-func NewQuicServer(ctx context.Context, cnf config.QuicTransport) (*QuicServer, error) {
+// NewServer creates a new QuicServer instance
+func NewServer(ctx context.Context, cnf config.QuicTransport) (*Server, error) {
 	tlsConfig, tcErr := cnf.GetTLSConfig()
 	if tcErr != nil {
 		return nil, errors.Wrapf(tcErr, "could not get TLS config for quic transport")
 	}
 
-	server := &QuicServer{
+	server := &Server{
 		ctx:             ctx,
-		handlerRegistry: make(map[HandlerType]QuicHandler),
+		handlerRegistry: make(map[types.HandlerType]QuicHandler),
 		cnf:             cnf,
 		tlsConfig:       tlsConfig,
 		stopChan:        make(chan struct{}),
@@ -48,12 +50,12 @@ func NewQuicServer(ctx context.Context, cnf config.QuicTransport) (*QuicServer,
 }
 
 // Addr returns the server address as a string.
-func (s *QuicServer) Addr() string {
+func (s *Server) Addr() string {
 	return s.cnf.Addr()
 }
 
 // Start starts the QUIC server
-func (s *QuicServer) Start() error {
+func (s *Server) Start() error {
 	var err error
 	s.listener, err = quic.ListenAddr(s.cnf.Addr(), s.tlsConfig, nil)
 	if err != nil {
@@ -71,7 +73,7 @@ func (s *QuicServer) Start() error {
 }
 
 // acceptConnections continuously accepts new QUIC connections
-func (s *QuicServer) acceptConnections() {
+func (s *Server) acceptConnections() {
 	defer s.wg.Done()
 
 	for {
@@ -91,7 +93,7 @@ func (s *QuicServer) acceptConnections() {
 	}
 }
 
-func (s *QuicServer) handleConnection(conn quic.Connection) {
+func (s *Server) handleConnection(conn quic.Connection) {
 	defer s.wg.Done()
 
 	// Continuously accept and handle new streams on the connection
@@ -131,7 +133,7 @@ func isClosedNetworkConnectionError(err error) bool {
 }
 
 // handleStream handles incoming QUIC streams
-func (s *QuicServer) handleStream(conn quic.Connection, stream quic.Stream) {
+func (s *Server) handleStream(conn quic.Connection, stream quic.Stream) {
 	defer s.wg.Done()
 
 	// Continuously read from the stream until it's closed
@@ -164,7 +166,7 @@ func (s *QuicServer) handleStream(conn quic.Connection, stream quic.Stream) {
 		}
 
 		// Step 2: Decode the message from the buffer
-		message, err := Decode(buffer[:n])
+		message, err := messages.Decode(buffer[:n])
 		if err != nil {
 			log.Printf("Error decoding message: %v", err)
 			return
@@ -186,24 +188,24 @@ func (s *QuicServer) handleStream(conn quic.Connection, stream quic.Stream) {
 }
 
 // Stop stops the QUIC server
-func (s *QuicServer) Stop() {
+func (s *Server) Stop() {
 	close(s.stopChan)
 	s.listener.Close()
 	s.wg.Wait()
 }
 
 // WaitStarted returns a channel that is closed when the server has started
-func (s *QuicServer) WaitStarted() <-chan struct{} {
+func (s *Server) WaitStarted() <-chan struct{} {
 	return s.started
 }
 
 // parseActionType parses the action type from the first byte of the stream
-func (s *QuicServer) parseActionType(frame []byte) (HandlerType, error) {
+func (s *Server) parseActionType(frame []byte) (types.HandlerType, error) {
 	if len(frame) < 1 {
 		return 0, fmt.Errorf("invalid action: frame too short")
 	}
 
-	var actionType HandlerType
+	var actionType types.HandlerType
 	err := actionType.FromByte(frame[0])
 	if err != nil {
 		return 0, err
@@ -213,11 +215,11 @@ func (s *QuicServer) parseActionType(frame []byte) (HandlerType, error) {
 }
 
 // RegisterHandler registers a handler for a specific action
-func (s *QuicServer) RegisterHandler(actionType HandlerType, handler QuicHandler) {
+func (s *Server) RegisterHandler(actionType types.HandlerType, handler QuicHandler) {
 	s.handlerRegistry[actionType] = handler
 }
 
 // DeregisterHandler deregisters a handler for a specific action
-func (s *QuicServer) DeregisterHandler(actionType HandlerType) {
+func (s *Server) DeregisterHandler(actionType types.HandlerType) {
 	delete(s.handlerRegistry, actionType)
 }

quic_server_test.go → transports/quic/server_test.go

@@ -1,4 +1,4 @@
-package fdb
+package transport_quic
 
 import (
 	"context"