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

Author: unit-adm

README.md

@@ -33,7 +33,11 @@ Make use of the client by importing it in your Go client source code. For exampl
 
 Unitdb supports Get, Put, Delete operations. It also supports encryption, batch operations, and writing to wildcard topics. See [usage guide](https://github.com/unit-io/unitdb/tree/master/docs/usage.md). 
 
-Samples are available in the examples directory for reference.
+Samples are available in the examples directory for reference. To build unitdb client 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-go=$GOPATH/src/github.com/unit-io/unitdb-go
+```
 
 ## Clustering
 To bring up the Unitdb cluster start 2 or more nodes. For fault tolerance 3 nodes or more are recommended.

docs/utp.md

@@ -24,6 +24,7 @@
  + [Message Types](#Message-Types)
    - [CONNECT](#CONNECT---Connection-Request)
    - [PUBLISH](#PUBLISH---Publish-message)
+   - [RELAY](#RELAY---Relay-request)
    - [SUBSCRIBE](#SUBSCRIBE---Subscribe-request)
    - [UNSUBSCRIBE](#UNSUBSCRIBE---Unsubscribe-request)
    - [PINGREQ](#PINGREQ---PING-request)
@@ -44,13 +45,15 @@ An application transport the data by uTP across network, it contains payload dat
 ### Client
 A Client opens the network connection to the Server using TCP/IP, WebSocket, GRPC or other bi-direction network protocols.
 - Pubslihes Application Mesasges to a topic that other Clients subscribes to.
+- Sends a relay request to retrieve persisted Application Messages from Server.
 - Subscribes to a topic to receive Application Messages.
 - Unsubcribe to remove a topic subscription.
 - Closes the network connection to the Server.
 
 ### Server
 - Accepts the network connections from Clients.
 - Recieves and store Application Messages published by Clients.
+- Relays pesisted Application Messages that matches the relay requests from Clients.
 - Processes topic subscription requests from Clients.
 - Route Application Messages that match Client subscriptions.
 - Closes the network connection from the Client.
@@ -75,24 +78,25 @@ Represented as enum value, the values are shown below.
 | RERSERVED | 0 | Forbidden |
 | CONNECT |	1 |	Client to Server |
 | PUBLISH |	2 |	Client to Server or Server to Client |
-| SUBSCRIBE | 3 | Client to Server |
-| UNSUBSCRIBE |	4	| Client to Server |
-| PINGREQ |	5 | Client to Server |
-| DISCONNECT | 6 | Client to Server or Server to Client |
+| RELAY | 3 | Client to Server |
+| SUBSCRIBE | 4 | Client to Server |
+| UNSUBSCRIBE |	5	| Client to Server |
+| PINGREQ |	6 | Client to Server |
+| DISCONNECT | 7 | Client to Server or Server to Client |
 
 ### Flow Control
 Flow Control is Control Message sent in response to a uTP Message Type or another Control Message.
 
-Client will send one of Message Type CONNECT, SUBSCRIBE, UNSUBSCRIBE, or PINGREQ  then Server will reponse with ACKNOWLEDGE Control Message. PUBLISH Message is sent from a Client to a Server or from a Server to a Client and if Delivery Mode is Express Delivery  then the receiver will respond with ACKNOWLEDGE Control Message. 
+Client will send one of Message Type CONNECT, RELAY, SUBSCRIBE, UNSUBSCRIBE, or PINGREQ  then Server will reponse with ACKNOWLEDGE Control Message. PUBLISH Message is sent from a Client to a Server or from a Server to a Client and if Delivery Mode is Express Delivery then the receiver will respond with ACKNOWLEDGE Control Message. 
 
 | Name | Value | Direction of Flow |
 | :--- | :--- | :--- |
 | None | 0 | None |
 | ACKNOWLEDGE | 1 | Client to Server or Server to Client |
 | NOTIFY | 2 | Server to Client |
 | RECEIVE | 3 | Client to Server |
-| RECEIPT | 4 | Client to Server or Server to Client |
-| COMPLETE | 5 | Client to Server or Server to Client |
+| RECEIPT | 4 | Client to Server |
+| COMPLETE | 5 | Server to Client |
 
 ### Message Length
 The Message Length represents number of bytes within the current Message.
@@ -106,7 +110,8 @@ In order to implement a Reliable Message Delivery flow the Client and Server nee
 The Server persists an unique Session State per Client Identifier. Client can also specify Sesson Key during Connection to persist multiple Session States per Client Identifier. The Session can continue after network re-connection until Client specify Clean Session during a new Connection.
 
 The Client Session pesist following States:
-- Reliable and Batch Delivery messages sent to the Server, but have not been completely acknowledged.
+- Reliable and Batch Delivery messages sent to the Server, but have not been acknowledged.
+- Relay request sent to the Server, but have not been acknowledged.
 - Reliable and Batch Delivery messages which have been received from the Server, but have not been completely acknowledged.
 
 The Server Session persiste following states:
@@ -278,6 +283,26 @@ The PublishMessage contains folowing fields:
 ### TTL
 A publisher can specify time to-live (TTL) when publishing an Application Message.
 
+### RELAY - Relay request
+The RELAY Message is sent from the Client to the Server to get persisted Application Messages from server for one or more topics. Each Relay request pairs the topics with last durations. The Server sends PUBLISH Messages to the Client to forward Application Messages that were persisted by the Server for the Topics that match these Relay requests. The RELAY Message also specifies (for each request) the Last duration for which the Server can send persisted Application Messages to the Client.
+
+The payload contains following fields.
+
+|  Name | Type |
+| :--- | :--- |
+| MessageID | int32 |
+| RelayRequests | repeated RelayRequest |
+
+The RelayRequest contains folowing fields:
+
+|  Name | Type |
+| :--- | :--- |
+| Topic | string |
+| Last | string |
+
+#### Last
+A Client can specify Last duration (for example "1h") to retrive persisted Application Messages published to the Topic.
+
 ### SUBSCRIBE - Subscribe request
 The SUBSCRIBE Message is sent from the Client to the Server to create one or more Subscriptions. Each Subscription registers one or more Topics for a Client. The Server sends PUBLISH Messages to the Client to forward Application Messages that were published to Topics that match these Subscriptions. The SUBSCRIBE Message also specifies (for each Subscription) the Delivery Mode with which the Server can send Application Messages to the Client.
 
@@ -295,10 +320,7 @@ The Subscription contains folowing fields:
 | DeliveryMode | int32 |
 | Delay | int32 |
 | Topic | string |
-| Last | string |
 
-#### Last
-A Subscriber can spcify Last duration (for example "1h") to retrive persisted Application Messages published to the subscribing Topic.
 
 ### UNSUBSCRIBE - Unsubscribe request
 An UNSUBSCRIBE Message is sent by the Client to the Server, to unsubscribe from topics.
@@ -320,7 +342,7 @@ The DISCONNECT Message is the final uTP Message sent from the Client or the Serv
 
 ## Control Message
 ### ACKNOWLEDGE - acknowledgement
-The ACKNOWLEDGE Control Message is sent by the Server in response to CONNECT, SUBSCRIBE, UNSUBSCRIBE, PUBLISH or PINGREQ Message from a Client. 
+The ACKNOWLEDGE Control Message is sent by the Server in response to CONNECT, RELAY, SUBSCRIBE, UNSUBSCRIBE, PUBLISH or PINGREQ Message from a Client. 
 
 The ACKNOWLEGE Message on CONNECT is sent with Message in the form of binary data that contains below Payload inforation. 
 
@@ -358,11 +380,9 @@ The Server must send a NOTIFY Control Message containing a new Message Identifie
 The Client must send a RECEIVE Control Message to respond to the NOTIFY Control Message to tell Server it is ready to receive the message. It must contain the same Message Identifier from NOTIFY Control Message.
 
 ### RECEIPT - Publish receipt
-The Server must respond with RECEIPT Control Message if it receive a PUBLISH Message with Delivery Mode as RELIABLE or BATCH Delivery. 
-
 The Client must respond to a PUBLISH Message with a RECEIPT Control Message if it has a matching subscription with Delivery Mode RELIABLE or BATCH Delivery and sender has sent a PUBLISH Message with RELIABLE or BATCH Delivery Mode.
 
 The RECEIPT Control Message must contain the same Message Identifier from PUBLISH Message.
 
 ### COMPLETE - Publish complete
-The Client or the Server must respond with COMPLETE Control Message if it receives a RECEIPT Control Message to mark the Publish complete. The COMPLETE Control Message must contain the same Message Identifier from RECEIPT Control Message.
+The Server must respond with COMPLETE Control Message if it receives a RECEIPT Control Message from Client to mark the Publish complete. The COMPLETE Control Message must contain the same Message Identifier from RECEIPT Control Message.

examples/pubsub/main.go

@@ -18,13 +18,13 @@ func main() {
 	defer db.Close()
 
 	// Use Entry.WithPayload() method to bulk store messages as topic is parsed one time on first request.
-	topic := []byte("teams.alpha.ch1.u1")
+	topic := []byte("teams.alpha.ch1.r1")
 	entry := &unitdb.Entry{Topic: topic}
 	for j := 0; j < 50; j++ {
-		db.PutEntry(entry.WithPayload([]byte(fmt.Sprintf("msg for team alpha channel1 receiver1 #%2d", j))))
+		db.PutEntry(entry.WithPayload([]byte(fmt.Sprintf("msg for team alpha channel1 recipient1 #%2d", j))))
 	}
 
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.u1?last=1h")).WithLimit(100)); err == nil {
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.r1?last=1h")).WithLimit(100)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
@@ -33,42 +33,42 @@ func main() {
 	// Writing to single topic in a batch
 	err = db.Batch(func(b *unitdb.Batch, completed <-chan struct{}) error {
 		topic := []byte("teams.alpha.ch1.*?ttl=1h")
-		b.Put(topic, []byte("msg for team alpha channel1 all receivers"))
+		b.Put(topic, []byte("msg for team alpha channel1 all recipients"))
 		return nil
 	})
 	if err != nil {
 		log.Fatal(err)
 		return
 	}
 
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.u2?last=1h")).WithLimit(10)); err == nil {
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.r2?last=1h")).WithLimit(10)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
 	}
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.u3?last=1h")).WithLimit(10)); err == nil {
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.r3?last=1h")).WithLimit(10)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
 	}
 
 	// Writing to multiple topics in a batch
 	err = db.Batch(func(b *unitdb.Batch, completed <-chan struct{}) error {
-		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.ch1.u2"), []byte("msg for team alpha channel1 receiver2")))
-		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.ch1.u3"), []byte("msg for team alpha channel1 receiver3")))
+		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.ch1.r2"), []byte("msg for team alpha channel1 recipient2")))
+		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.ch1.r3"), []byte("msg for team alpha channel1 recipient3")))
 		return nil
 	})
 	if err != nil {
 		log.Fatal(err)
 		return
 	}
 
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.u2?last=1h")).WithLimit(10)); err == nil {
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.r2?last=1h")).WithLimit(10)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
 	}
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.u3?last=1h")).WithLimit(10)); err == nil {
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.alpha.ch1.r3?last=1h")).WithLimit(10)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
@@ -82,19 +82,19 @@ func main() {
 	err = db.Batch(func(b *unitdb.Batch, completed <-chan struct{}) error {
 		b.SetOptions(unitdb.WithBatchContract(contract))
 		topic := []byte("teams.alpha.ch1.*?ttl=1h")
-		b.Put(topic, []byte("msg for team alpha channel1 all receivers #1"))
-		b.Put(topic, []byte("msg for team alpha channel1 all receivers #2"))
-		b.Put(topic, []byte("msg for team alpha channel1 all receivers #3"))
+		b.Put(topic, []byte("msg for team alpha channel1 all recipients #1"))
+		b.Put(topic, []byte("msg for team alpha channel1 all recipients #2"))
+		b.Put(topic, []byte("msg for team alpha channel1 all recipients #3"))
 		return nil
 	})
 
 	// Writing to multiple topics in a batch
 	err = db.Batch(func(b *unitdb.Batch, completed <-chan struct{}) error {
 		b.SetOptions(unitdb.WithBatchContract(contract))
-		b.PutEntry(unitdb.NewEntry([]byte("teams.*.ch1"), []byte("msg for any team channel1")))
-		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.*"), []byte("msg for team alpha all channels")))
+		b.PutEntry(unitdb.NewEntry([]byte("teams.*.ch1"), []byte("msg for channel1 in any teams")))
+		b.PutEntry(unitdb.NewEntry([]byte("teams.alpha.*"), []byte("msg for all channels in team alpha")))
 		b.PutEntry(unitdb.NewEntry([]byte("teams..."), []byte("msg for all teams and all channels")))
-		b.PutEntry(unitdb.NewEntry([]byte("..."), []byte("msg broadcast to all receivers of all teams all channels")))
+		b.PutEntry(unitdb.NewEntry([]byte("..."), []byte("msg broadcast to all recipients of all channels in all teams")))
 		return nil
 	})
 	if err != nil {
@@ -116,8 +116,8 @@ func main() {
 		}
 	}
 
-	// Get message for team beta channel2 receiver1
-	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.beta.ch2.u1?last=1h")).WithLimit(10)); err == nil {
+	// Get message for team beta channel2 recipient11
+	if msgs, err := db.Get(unitdb.NewQuery([]byte("teams.beta.ch2.r1?last=1h")).WithLimit(10)); err == nil {
 		for _, msg := range msgs {
 			log.Printf("%s ", msg)
 		}
@@ -127,8 +127,8 @@ func main() {
 	// Note, encryption can also be set on entire database using DB.Open() and set encryption flag in options parameter.
 	err = db.Batch(func(b *unitdb.Batch, completed <-chan struct{}) error {
 		b.SetOptions(unitdb.WithBatchEncryption())
-		topic := []byte("teams.alpha.ch1.u1?ttl=1h")
-		b.Put(topic, []byte("msg for team alpha channel1 receiver1"))
+		topic := []byte("teams.alpha.ch1.r1?ttl=1h")
+		b.Put(topic, []byte("msg for team alpha channel1 recipient1"))
 		return nil
 	})