[server] add suport for more transport protocols

Author: unit-adm

examples/pubsub/main.go

@@ -16,7 +16,10 @@ var f unitdb.MessageHandler = func(client unitdb.Client, msg unitdb.Message) {
 }
 
 func main() {
+	// Create a new client to open a network connection using the protocol indicated in the URL.
 	client, err := unitdb.NewClient(
+		//"tcp://localhost:6060",
+		// "ws://localhost:6080",
 		"grpc://localhost:6080",
 		"UCBFDONCNJLaKMCAIeJBaOVfbAXUZHNPLDKKLDKLHZHKYIZLCDPQ",
 		unitdb.WithInsecure(),

server/internal/cluster.go

@@ -307,7 +307,7 @@ func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
 			}
 
 			log.Info("cluster.Master", "new connection request"+string(msg.Conn.ConnID))
-			conn = Globals.Service.newRpcConn(node, msg.Conn.Proto, msg.Conn.ConnID, msg.Conn.ClientID)
+			conn = Globals.Service.newRpcConn(node, msg.Conn.ConnID, msg.Conn.ClientID)
 			go conn.rpcWriteLoop()
 		}
 		// Update session params which may have changed since the last call.
@@ -374,7 +374,7 @@ func (c *Cluster) isRemoteContract(contract string) bool {
 }
 
 // Forward client message to the Master (cluster node which owns the topic)
-func (c *Cluster) routeToContract(msg lp.Packet, topic *security.Topic, msgType uint8, m *message.Message, conn *_Conn) error {
+func (c *Cluster) routeToContract(msg lp.LineProtocol, topic *security.Topic, msgType uint8, m *message.Message, conn *_Conn) error {
 	// Find the cluster node which owns the topic, then forward to it.
 	n := c.nodeForContract(string(conn.clientid.Contract()))
 	if n == nil {

server/internal/conn.go

@@ -29,7 +29,7 @@ import (
 	"github.com/unit-io/unitdb/server/internal/message"
 	"github.com/unit-io/unitdb/server/internal/message/security"
 	lp "github.com/unit-io/unitdb/server/internal/net"
-	"github.com/unit-io/unitdb/server/internal/net/grpc"
+	"github.com/unit-io/unitdb/server/internal/net/pubsub"
 	"github.com/unit-io/unitdb/server/internal/pkg/log"
 	"github.com/unit-io/unitdb/server/internal/pkg/uid"
 	"github.com/unit-io/unitdb/server/internal/store"
@@ -44,8 +44,8 @@ type _Conn struct {
 	adp    lp.ProtoAdapter
 	socket net.Conn
 	// send     chan []byte
-	send               chan lp.Packet
-	recv               chan lp.Packet
+	send               chan lp.LineProtocol
+	recv               chan lp.LineProtocol
 	pub                chan *lp.Publish
 	stop               chan interface{}
 	insecure           bool           // The insecure flag provided by client will not perform key validation and permissions check on the topic.
@@ -65,21 +65,13 @@ type _Conn struct {
 	closeC chan struct{}
 }
 
-func (s *_Service) newConn(t net.Conn, proto lp.Proto) *_Conn {
-	var adp lp.ProtoAdapter
-	switch proto {
-	case lp.GRPC:
-		adp = &grpc.LineProto{}
-	case lp.GRPC_WEB:
-		adp = &grpc.LineProto{}
-	}
-
+func (s *_Service) newConn(t net.Conn) *_Conn {
 	c := &_Conn{
-		adp:        adp,
+		adp:        &pubsub.Packet{},
 		socket:     t,
 		MessageIds: message.NewMessageIds(),
-		send:       make(chan lp.Packet, 1), // buffered
-		recv:       make(chan lp.Packet),
+		send:       make(chan lp.LineProtocol, 1), // buffered
+		recv:       make(chan lp.LineProtocol),
 		pub:        make(chan *lp.Publish),
 		stop:       make(chan interface{}, 1), // Buffered by 1 just to make it non-blocking
 		connid:     uid.NewLID(),
@@ -97,21 +89,14 @@ func (s *_Service) newConn(t net.Conn, proto lp.Proto) *_Conn {
 }
 
 // newRpcConn a new connection in cluster
-func (s *_Service) newRpcConn(conn interface{}, proto lp.Proto, connid uid.LID, clientid uid.ID) *_Conn {
-	var adp lp.ProtoAdapter
-	switch proto {
-	case lp.GRPC:
-		adp = &grpc.LineProto{}
-	case lp.GRPC_WEB:
-		adp = &grpc.LineProto{}
-	}
+func (s *_Service) newRpcConn(conn interface{}, connid uid.LID, clientid uid.ID) *_Conn {
 	c := &_Conn{
-		adp:        adp,
+		adp:        &pubsub.Packet{},
 		connid:     connid,
 		clientid:   clientid,
 		MessageIds: message.NewMessageIds(),
-		send:       make(chan lp.Packet, 1), // buffered
-		recv:       make(chan lp.Packet),
+		send:       make(chan lp.LineProtocol, 1), // buffered
+		recv:       make(chan lp.LineProtocol),
 		pub:        make(chan *lp.Publish),
 		stop:       make(chan interface{}, 1), // Buffered by 1 just to make it non-blocking
 		service:    s,
@@ -311,7 +296,7 @@ func (c *_Conn) outboundID(mid message.MID) (id uint16) {
 	return uint16(uint32(c.connid) - (uint32(mid)))
 }
 
-func (c *_Conn) storeInbound(m lp.Packet) {
+func (c *_Conn) storeInbound(m lp.LineProtocol) {
 	if c.clientid != nil {
 		blockId := uint64(c.clientid.Contract())
 		k := uint64(c.inboundID(m.Info().MessageID))<<32 + blockId
@@ -320,7 +305,7 @@ func (c *_Conn) storeInbound(m lp.Packet) {
 	}
 }
 
-func (c *_Conn) storeOutbound(m lp.Packet) {
+func (c *_Conn) storeOutbound(m lp.LineProtocol) {
 	if c.clientid != nil {
 		blockId := uint64(c.clientid.Contract())
 		k := uint64(c.inboundID(m.Info().MessageID))<<32 + blockId

server/internal/hdl_conn.go

@@ -26,6 +26,7 @@ import (
 	"github.com/unit-io/unitdb/server/internal/message"
 	"github.com/unit-io/unitdb/server/internal/message/security"
 	lp "github.com/unit-io/unitdb/server/internal/net"
+	"github.com/unit-io/unitdb/server/internal/net/pubsub"
 	"github.com/unit-io/unitdb/server/internal/pkg/crypto"
 	"github.com/unit-io/unitdb/server/internal/pkg/log"
 	"github.com/unit-io/unitdb/server/internal/pkg/stats"
@@ -66,7 +67,7 @@ func (c *_Conn) readLoop() error {
 }
 
 // handle handles inbound packets.
-func (c *_Conn) handle(pkt lp.Packet) error {
+func (c *_Conn) handle(pkt lp.LineProtocol) error {
 	start := time.Now()
 	var status int = 200
 	defer func() {
@@ -83,9 +84,16 @@ func (c *_Conn) handle(pkt lp.Packet) error {
 		var returnCode uint8
 		packet := *pkt.(*lp.Connect)
 
+		// switch the proto adapter based on protoname in the connect packet.
+		switch packet.ProtoName {
+		case "TELEMETRY":
+			c.adp = &pubsub.Packet{}
+		case "INGESTION":
+		case "QUERY":
+		}
 		c.insecure = packet.InsecureFlag
 		c.username = string(packet.Username)
-		clientid, err := c.onConnect(packet.ClientID)
+		clientid, err := c.onConnect([]byte(packet.ClientID))
 		if err != nil {
 			status = err.Status
 			returnCode = 0x05 // Unauthorized

server/internal/net/hdl_grpc.go

@@ -72,7 +72,7 @@ func (s *GrpcServer) Stream(stream pbx.Unitdb_StreamServer) error {
 	conn := StreamConn(stream)
 	defer conn.Close()
 
-	go s.Handler(conn, GRPC)
+	go s.Handler(conn)
 	<-stream.Context().Done()
 	return nil
 }

server/internal/net/hdl_grpc_web.go

@@ -93,7 +93,7 @@ func (s *HttpServer) HandleFunc(w http.ResponseWriter, r *http.Request) {
 		return nil
 	})
 
-	go s.Handler(newConn(ws), GRPC_WEB)
+	go s.Handler(newConn(ws))
 }
 
 func (s *HttpServer) Serve(list net.Listener) error {

server/internal/net/hdl_tcp.go

@@ -67,6 +67,6 @@ func (s *TcpServer) Serve(list net.Listener) error {
 		}
 
 		tempDelay = 0
-		go s.Handler(conn, GRPC)
+		go s.Handler(conn)
 	}
 }