add pubsub relay to uTP protocol to retrieve persisted messages from server

Author: unit-adm

README.md

@@ -7,7 +7,7 @@ The Unitdb is a real-time messaging system for microservices, and real-tme inter
 ## Quick Start
 To build [unitdb](https://github.com/unit-io/unitdb) from source code use go get command and copy unitdb.conf to the path unitdb binary is placed.
 
-> go get -u github.com/unit-io/unitdb
+> go get -u github.com/unit-io/unitdb/server
 
 ### Usage
 Detailed API documentation is available using the [godoc.org](https://godoc.org/github.com/unit-io/unitdb-go) service.
@@ -16,7 +16,11 @@ Make use of the client by importing it in your Go client source code. For exampl
 
 import "github.com/unit-io/unitdb-go"
 
-Samples are available in the cmd directory for reference.
+Samples are available in the cmd directory for reference. To build unitdb server from latest source code use "replace" in go.mod to point to your local module.
+
+```golang
+go mod edit -replace github.com/unit-io/unitdb=$GOPATH/src/github.com/unit-io/unitdb
+```
 
 ## Contributing
 If you'd like to contribute, please fork the repository and use a feature branch. Pull requests are welcome.

batch.go

@@ -5,7 +5,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/unit-io/unitdb-go/utp"
+	"github.com/unit-io/unitdb-go/internal/utp"
 )
 
 const (

client.go

@@ -11,14 +11,14 @@ import (
 	"time"
 
 	"github.com/golang/protobuf/proto"
-	"github.com/unit-io/unitdb-go/store"
-	"github.com/unit-io/unitdb-go/utp"
+	"github.com/unit-io/unitdb-go/internal/store"
+	"github.com/unit-io/unitdb-go/internal/utp"
 	"github.com/unit-io/unitdb/server/common"
 	pbx "github.com/unit-io/unitdb/server/proto"
 	"google.golang.org/grpc"
 
 	// Database store
-	_ "github.com/unit-io/unitdb-go/db/unitdb"
+	_ "github.com/unit-io/unitdb-go/internal/db/unitdb"
 )
 
 type Client interface {
@@ -39,23 +39,24 @@ type Client interface {
 	Publish(topic string, payload []byte, pubOpts ...PubOptions) Result
 	// Subscribe starts a new subscription. Provide a MessageHandler to be executed when
 	// a message is published on the topic provided, or nil for the default handler.
+	// Relay sends a relay request to server. Provide a MessageHandler to be executed when
+	// a message is published on the topic provided, or nil for the default handler.
+	Relay(topic string, relOpts ...RelOptions) Result
 	Subscribe(topic string, subOpts ...SubOptions) Result
 	// Unsubscribe will end the subscription from each of the topics provided.
 	// Messages published to those topics from other clients will no longer be
 	// received.
 	Unsubscribe(topics ...string) Result
 }
 type client struct {
-	mu         sync.Mutex // mutex for the connection
 	opts       *options
 	context    context.Context    // context for the client
 	cancel     context.CancelFunc // cancellation function
 	messageIds                    // local identifier of messages
 	connID     int32              // The unique id of the connection.
-	sessID uint32
-	epoch    uint32             // The session ID of the connection.
-	conn       net.Conn           // the network connection
-	stream     grpc.Stream
+	sessID     uint32
+	epoch      uint32   // The session ID of the connection.
+	conn       net.Conn // the network connection
 	send       chan *MessageAndResult
 	recv       chan utp.Message
 	pub        chan *utp.Publish
@@ -180,18 +181,19 @@ func (c *client) ConnectContext(ctx context.Context) error {
 		c.updateLastTouched()
 		go c.keepalive(ctx)
 	}
+	// c.closeW.Add(3)
 	go c.readLoop(ctx)   // process incoming messages
 	go c.writeLoop(ctx)  // send messages to servers
 	go c.dispatcher(ctx) // dispatch messages to client
 
-		// Take care of any messages in the store
-		var sessKey uint32
-		if c.opts.sessionKey != 0 {
-			sessKey = c.opts.sessionKey
-		} else {
-			sessKey = c.epoch
-		}
-	
+	// Take care of any messages in the store
+	var sessKey uint32
+	if c.opts.sessionKey != 0 {
+		sessKey = c.opts.sessionKey
+	} else {
+		sessKey = c.epoch
+	}
+
 	sessID := c.connID
 	if rawSess, err := store.Session.Get(uint64(sessKey)); err == nil {
 		sessID := binary.LittleEndian.Uint32(rawSess[:4])
@@ -275,7 +277,7 @@ func (c *client) DisconnectContext(ctx context.Context) error {
 		// Disconnect() called but not connected
 		return nil
 	}
-	
+
 	defer c.close()
 	m := &utp.Disconnect{}
 	r := &DisconnectResult{result: result{complete: make(chan struct{})}}
@@ -354,6 +356,42 @@ func (c *client) Publish(topic string, payload []byte, pubOpts ...PubOptions) Re
 	return r
 }
 
+// Relay send a new relay request. Provide a MessageHandler to be executed when
+// a message is published on the topic provided.
+func (c *client) Relay(topic string, relOpts ...RelOptions) Result {
+	r := &RelayResult{result: result{complete: make(chan struct{})}}
+	if err := c.ok(); err != nil {
+		r.setError(errors.New("error not connected"))
+		return r
+	}
+	opts := new(relOptions)
+	for _, opt := range relOpts {
+		opt.set(opts)
+	}
+
+	rel := &utp.Relay{}
+	rel.RelayRequests = append(rel.RelayRequests, &utp.RelayRequest{Topic: topic, Last: opts.last})
+
+	if rel.MessageID == 0 {
+		mID := c.nextID(r)
+		rel.MessageID = c.outboundID(mID)
+	}
+	relayWaitTimeout := c.opts.writeTimeout
+	if relayWaitTimeout == 0 {
+		relayWaitTimeout = time.Second * 30
+	}
+	// persist outbound
+	c.storeOutbound(rel)
+	select {
+	case c.send <- &MessageAndResult{m: rel, r: r}:
+	case <-time.After(relayWaitTimeout):
+		r.setError(errors.New("relay request timeout error occurred"))
+		return r
+	}
+
+	return r
+}
+
 // Subscribe starts a new subscription. Provide a MessageHandler to be executed when
 // a message is published on the topic provided.
 func (c *client) Subscribe(topic string, subOpts ...SubOptions) Result {
@@ -368,10 +406,7 @@ func (c *client) Subscribe(topic string, subOpts ...SubOptions) Result {
 	}
 
 	sub := &utp.Subscribe{}
-	sub.Subscriptions = append(sub.Subscriptions, &utp.Subscription{DeliveryMode: opts.deliveryMode, Delay: opts.delay, Topic: topic, Last: opts.last})
-
-	if opts.callback != nil {
-	}
+	sub.Subscriptions = append(sub.Subscriptions, &utp.Subscription{DeliveryMode: opts.deliveryMode, Delay: opts.delay, Topic: topic})
 
 	if sub.MessageID == 0 {
 		mID := c.nextID(r)
@@ -435,20 +470,35 @@ func (c *client) resume(prefix uint32, subscription bool) {
 		// isKeyOutbound
 		if (k & (1 << 4)) == 0 {
 			switch msg.Type() {
+			case utp.RELAY:
+				p := msg.(*utp.Relay)
+				r := &RelayResult{result: result{complete: make(chan struct{})}}
+				r.messageID = msg.Info().MessageID
+				c.messageIds.resumeID(MID(r.messageID))
+				// var topics []RelayRequest
+				// for _, req := range p.RelayRequests {
+				// 	var t RelayRequest
+				// 	t.Topic = req.Topic
+				// 	t.Last = req.Last
+				// 	topics = append(topics, t)
+				// }
+				r.reqs = p.RelayRequests
+				c.send <- &MessageAndResult{m: msg, r: r}
 			case utp.SUBSCRIBE:
 				if subscription {
 					p := msg.(*utp.Subscribe)
 					r := &SubscribeResult{result: result{complete: make(chan struct{})}}
 					r.messageID = msg.Info().MessageID
 					c.messageIds.resumeID(MID(r.messageID))
-					var topics []Subscription
-					for _, sub := range p.Subscriptions {
-						var t Subscription
-						t.Topic = sub.Topic
-						t.DeliveryMode = sub.DeliveryMode
-						topics = append(topics, t)
-					}
-					r.subs = append(r.subs, topics...)
+					// var topics []Subscription
+					// for _, sub := range p.Subscriptions {
+					// 	var t Subscription
+					// 	t.Topic = sub.Topic
+					// 	t.DeliveryMode = sub.DeliveryMode
+					// 	topics = append(topics, t)
+					// }
+					// r.subs = append(r.subs, topics...)
+					r.subs = p.Subscriptions
 					c.send <- &MessageAndResult{m: msg, r: r}
 				}
 			case utp.UNSUBSCRIBE:

cmd/sample/main.go

@@ -28,17 +28,17 @@ Options:
 */
 
 func main() {
-	topic := flag.String("topic", "AbYANcEGXRTLC/teams.alpha.user1", "The topic name to/from which to publish/subscribe")
+	topic := flag.String("topic", "DDcBReFJBDFKe/groups.private.673651407196578720.message", "The topic name to/from which to publish/subscribe")
 	server := flag.String("server", "grpc://localhost:6080", "The server URI. ex: grpc://127.0.0.1:6080")
 	password := flag.String("password", "", "The password (optional)")
 	user := flag.String("user", "", "The User (optional)")
 	id := flag.String("id", "UCBFDONCNJLaKMCAIeJBaOVfbAXUZHNPLDKKLDKLHZHKYIZLCDPQ", "The ClientID (optional)")
 	num := flag.Int("number", 1, "The number of messages to publish or subscribe (default 1)")
 	payload := flag.String("message", "Hello team alpha channel1!", "The message text to publish (default empty)")
-	action := flag.String("action", "sub", "Action publish or subscribe (required)")
+	action := flag.String("action", "relay", "Action publish, relay or subscribe (required)")
 	flag.Parse()
 
-	if *action != "pub" && *action != "sub" && *action != "unsub" && *action != "keygen" {
+	if *action != "pub" && *action != "relay" && *action != "sub" && *action != "unsub" && *action != "keygen" {
 		fmt.Println("Invalid setting for -action, must be pub or sub")
 		return
 	}
@@ -86,12 +86,13 @@ func main() {
 			log.Fatalf("err: %s", err)
 		}
 		fmt.Println("Keygen Started")
-		req := struct {
+		req := []struct {
 			Topic string `json:"topic"`
 			Type  string `json:"type"`
-		}{
+		}{{
 			*topic,
 			"rw",
+		},
 		}
 		keyReq, err := json.Marshal(req)
 		if err != nil {
@@ -114,7 +115,57 @@ func main() {
 		}
 	}
 
+	if *action == "sub" {
+		recv := make(chan [2][]byte)
+
+		client, err := unitdb.NewClient(
+			*server,
+			*id,
+			// unitdb.WithInsecure(),
+			unitdb.WithSessionKey(2339641922),
+			unitdb.WithUserNamePassword(*user, []byte(*password)),
+			// unitdb.WithCleanSession(),
+			unitdb.WithKeepAlive(2*time.Second),
+			unitdb.WithPingTimeout(1*time.Second),
+			unitdb.WithConnectionLostHandler(func(client unitdb.Client, err error) {
+				if err != nil {
+					log.Fatal(err)
+				}
+				close(recv)
+			}),
+			unitdb.WithDefaultMessageHandler(func(client unitdb.Client, msg unitdb.Message) {
+				recv <- [2][]byte{[]byte(msg.Topic()), msg.Payload()}
+			}),
+			unitdb.WithBatchDuration(10*time.Second),
+		)
+		if err != nil {
+			log.Fatalf("err: %s", err)
+		}
+		ctx := context.Background()
+		err = client.ConnectContext(ctx)
+		if err != nil {
+			log.Fatalf("err: %s", err)
+		}
+		r := client.Subscribe(*topic, unitdb.WithSubDeliveryMode(1) /*, unitdb.WithSubDelay(1*time.Second)*/)
+		if _, err := r.Get(ctx, 1*time.Second); err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+
+		for {
+			select {
+			case <-ctx.Done():
+				client.DisconnectContext(ctx)
+				fmt.Println("Subscriber Disconnected")
+				return
+			case incoming := <-recv:
+				fmt.Printf("RECEIVED TOPIC: %s MESSAGE: %s\n", incoming[0], incoming[1])
+			}
+		}
+	}
+
 	if *action == "pub" {
+
 		client, err := unitdb.NewClient(
 			*server,
 			*id,
@@ -147,6 +198,7 @@ func main() {
 		time.Sleep(1 * time.Second)
 		client.DisconnectContext(ctx)
 		fmt.Println("Publisher Disconnected")
+
 	} else {
 		recv := make(chan [2][]byte)
 
@@ -178,7 +230,7 @@ func main() {
 		if err != nil {
 			log.Fatalf("err: %s", err)
 		}
-		r := client.Subscribe(*topic, unitdb.WithSubDeliveryMode(1)/*, unitdb.WithSubDelay(1*time.Second)*/)
+		r := client.Relay(*topic, unitdb.WithLast("1m"))
 		if _, err := r.Get(ctx, 1*time.Second); err != nil {
 			fmt.Println(err)
 			os.Exit(1)
@@ -188,10 +240,13 @@ func main() {
 			select {
 			case <-ctx.Done():
 				client.DisconnectContext(ctx)
-				fmt.Println("Subscriber Disconnected")
+				fmt.Println("Client Disconnected")
 				return
 			case incoming := <-recv:
 				fmt.Printf("RECEIVED TOPIC: %s MESSAGE: %s\n", incoming[0], incoming[1])
+				client.DisconnectContext(ctx)
+				fmt.Println("Client Disconnected")
+				return
 			}
 		}
 	}

cmd/simple/main.go

@@ -37,23 +37,31 @@ func main() {
 		log.Fatalf("err: %s", err)
 	}
 
-	r := client.Subscribe("teams.alpha.user1", unitdb.WithLast("1m"), unitdb.WithSubDeliveryMode(0))
+	var r unitdb.Result
+
+	r = client.Relay("ADcABeFRBDJKe/groups.private.673651407196578720.message", unitdb.WithLast("10m"))
+	if _, err := r.Get(ctx, 1*time.Second); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+
+	r = client.Subscribe("ADcABeFRBDJKe/groups.private.673651407196578720.message", unitdb.WithSubDeliveryMode(0))
 	if _, err := r.Get(ctx, 1*time.Second); err != nil {
 		fmt.Println(err)
 		os.Exit(1)
 	}
 
 	for i := 0; i < 2; i++ {
 		msg := fmt.Sprintf("Hi #%d time!", i)
-		r := client.Publish("teams.alpha.user1", []byte(msg), unitdb.WithTTL("1m"), unitdb.WithPubDeliveryMode(0))
+		r := client.Publish("ADcABeFRBDJKe/groups.private.673651407196578720.message", []byte(msg), unitdb.WithTTL("1m"), unitdb.WithPubDeliveryMode(0))
 		if _, err := r.Get(ctx, 1*time.Second); err != nil {
 			log.Fatalf("err: %s", err)
 		}
 	}
 
 	wait := time.NewTicker(5 * time.Second)
 	<-wait.C
-	r = client.Unsubscribe("teams.alpha.user1")
+	r = client.Unsubscribe("ADcABeFRBDJKe/groups.private.673651407196578720.message")
 	if _, err := r.Get(ctx, 1*time.Second); err != nil {
 		fmt.Println(err)
 		os.Exit(1)

go.mod

@@ -4,7 +4,8 @@ go 1.16
 
 require (
 	github.com/golang/protobuf v1.5.2
-	github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264 // indirect
-	github.com/unit-io/unitdb v0.1.0
-	google.golang.org/grpc v1.37.0
+	github.com/unit-io/unitdb v0.1.1
+	google.golang.org/grpc v1.39.0
 )
+
+replace github.com/unit-io/unitdb => /src/github.com/unit-io/unitdb