[server] add relay to retrieve messages persisted on unitdb server

Author: unit-adm

README.md

@@ -33,11 +33,7 @@ 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. 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
-```
+Samples are available in the examples directory for reference.
 
 ## Clustering
 To bring up the Unitdb cluster start 2 or more nodes. For fault tolerance 3 nodes or more are recommended.
@@ -49,6 +45,11 @@ To bring up the Unitdb cluster start 2 or more nodes. For fault tolerance 3 node
 
 Above example shows each Unitdb node running on the same host, so each node must listen on different ports. This would not be necessary if each node ran on a different host.
 
+## Client Libraries
+Make use of officially supported client libraries to connect to unitdb server running on single node and running on a cluster.
+- [unitdb-go](https://github.com/unit-io/unitdb-go) Lightweight and high performance unitdb Go client library.
+- [unitdb-dart](https://github.com/unit-io/unitdb-dart) High performance unitdb Flutter/Dart client library.
+
 ## Architecture Overview
 The unitdb engine handles data from the point put request is received through writing data to the physical disk. Data is compressed and encrypted (if encryption is set) then written to a WAL for immediate durability. Entries are written to memdb and become immediately queryable. The memdb entries are periodically written to log files in the form of blocks.
 

docs/utp.md

@@ -45,7 +45,7 @@ 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.
+- Sends a relay request to retrieve Application Messages persisted on Server.
 - Subscribes to a topic to receive Application Messages.
 - Unsubcribe to remove a topic subscription.
 - Closes the network connection to the Server.
@@ -284,7 +284,7 @@ The PublishMessage contains folowing fields:
 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 RELAY Message is sent from the Client to the Server to get Application Messages persisted on 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 stored Application Messages to the Client.
 
 The payload contains following fields.
 
@@ -301,7 +301,7 @@ The RelayRequest contains folowing fields:
 | Last | string |
 
 #### Last
-A Client can specify Last duration (for example "1h") to retrive persisted Application Messages published to the Topic.
+A Client can specify Last duration (for example "1h") to retrive 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.

examples/server_client/main.go

@@ -9,10 +9,7 @@ require (
 	github.com/gorilla/websocket v1.4.2
 	github.com/rs/zerolog v1.21.0
 	github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264
-	github.com/unit-io/unitdb-go v0.0.0-20210407101657-d9db9270f78d
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
 	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
 	google.golang.org/grpc v1.39.0
 )
-
-// replace github.com/unit-io/unitdb-go => /src/github.com/unit-io/unitdb-go

go.mod

@@ -15,7 +15,6 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -59,7 +58,6 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264 h1:31MK/k8NNVPI3UEUCJ8+9aEzlFxJRhawKk2gqxSbJeA=
 github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264/go.mod h1:jLqAtkF257MDiAc5K8svPVUGjfig2qdIhnWs3OCDwKg=
-github.com/unit-io/unitdb v0.1.1/go.mod h1:B8CPnLiFt1OcRxWIj+69kNxMRuSoS7o3muBzgqLZbkk=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -129,7 +127,6 @@ google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQ
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTpR3n0=
 google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.37.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.39.0 h1:Klz8I9kdtkIN6EpHHUOMLCYhTn/2WAe5a0s1hcBkdTI=
 google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=

go.sum

@@ -23,7 +23,7 @@ Make use of the client by importing it in your Go client source code. For exampl
 
 The memdb supports Get, Set, Delete operations. It also supports batch operations.
 
-Samples are available in the cmd directory for reference.
+Samples are available in the examples directory for reference.
 
 ### Opening a database
 To open or create a new database, use the memdb.Open() function:
@@ -101,7 +101,7 @@ use DB.Delete() function to delete a key-value pair.
 ```
 
 ### Batch operation
-Use batch operation to bulk insert records into memdb or bulk delete records from memdb. See examples under cmd/memdb folder.
+Use batch operation to bulk insert records into memdb or bulk delete records from memdb. See examples under examples/memdb folder.
 
 #### Writing to a batch
 Use Batch.Put() to insert a new key-value or Batch.Delete() to delete a key-value from DB.

memdb/README.md

@@ -33,10 +33,10 @@ const (
 	TopicInvalid = uint8(iota)
 	TopicStatic
 	TopicWildcard
-	TopicWildcardSymbol = '*'
-	TopicGenericSymbol  = "..."
-	TopicSeparator      = '.' // The separator character.
-	TopicMaxDepth       = 100 // Maximum depth for topic using a separator
+	TopicWildcardSymbol      = '*'
+	TopicMultiWildcardSymbol = "..."
+	TopicSeparator           = '.' // The separator character.
+	TopicMaxDepth            = 100 // Maximum depth for topic using a separator
 
 	// Wildcard wildcard is hash for wildcard topic such as '*' or '...'
 	Wildcard = uint32(857445537)
@@ -132,15 +132,6 @@ func (t *Topic) Last() (time.Time, int, bool) {
 	return zeroTime, 0, ok
 }
 
-// toUnix converts the time to Unix Time with validation.
-func toUnix(t int64) time.Time {
-	if t == 0 {
-		return zeroTime
-	}
-
-	return time.Unix(t, 0)
-}
-
 // getOption retrieves a Uint option.
 func (t *Topic) getOption(name string) (string, int, bool) {
 	for i := 0; i < len(t.Options); i++ {
@@ -207,8 +198,6 @@ func (t *Topic) Parse(contract uint32, wildcard bool) {
 		return
 	}
 	parseStaticTopic(contract, t)
-
-	return
 }
 
 // parseStaticTopic attempts to parse the topic from the underlying slice.
@@ -254,10 +243,10 @@ func parseWildcardTopic(contract uint32, topic *Topic) (ok bool) {
 	}
 
 	depth := uint8(0)
-	q := []byte(TopicGenericSymbol)
+	q := []byte(TopicMultiWildcardSymbol)
 	if bytes.HasSuffix(topic.Topic, q) {
 		depth++
-		topic.Topic = bytes.TrimRight(topic.Topic, string(TopicGenericSymbol))
+		topic.Topic = bytes.TrimRight(topic.Topic, string(TopicMultiWildcardSymbol))
 		topic.TopicType = TopicWildcard
 		topic.Depth = TopicMaxDepth
 	}

message/topic.go

@@ -135,7 +135,7 @@ func (c *_Conn) SendMessage(msg *message.Message) bool {
 	}
 
 	// Check batch, relay or delay delivery.
-	if msg.DeliveryMode == 2 || msg.DeliveryMode == 3 || msg.Delay > 0 {
+	if msg.DeliveryMode == 2 || msg.Delay > 0 {
 		c.batchManager.add(msg.Delay, pubMsg)
 		return true
 	}
@@ -180,7 +180,7 @@ func (c *_Conn) subscribe(subMsg utp.Subscribe, topic *security.Topic, sub *utp.
 
 	key := string(topic.Key)
 	if key == "" {
-		key, err = security.GenerateKey(c.clientID.Contract(), []byte(topic.Topic), security.AllowNone)
+		key, err = security.GenerateKey(c.clientID.Contract(), topic.Topic, security.AllowNone)
 		if err != nil {
 			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
 			return err
@@ -302,7 +302,7 @@ func (c *_Conn) publish(pub utp.Publish, topic *security.Topic, pubMsg *utp.Publ
 
 	if !pub.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
 		if err = Globals.Cluster.routeToContract(&pub, topic, message.PUBLISH, msg, c); err != nil {
-			log.ErrLogger.Err(err).Str("context", "conn.publish").Int64("connid", int64(c.connID)).Msg("unable to publish to remote topic")
+			log.ErrLogger.Err(err).Str("context", "conn.publish").Int64("connid", int64(c.connID)).Msg("unable to publish to a remote topic")
 			return err
 		}
 	}

server/internal/conn.go

@@ -44,25 +44,25 @@ type Adapter interface {
 	// SSID, where first element should be a contract ID. The time resolution
 	// for TTL will be in seconds. The function is executed synchronously and
 	// it returns an error if some error was encountered during storage.
-	Put(contract uint32, topic, payload []byte, ttl string) error
+	Put(contract uint32, topic string, payload []byte, ttl string) error
 
 	// PutWithID is used to store a message using a pre generated ID, the SSID provided must be a full SSID
 	// SSID, where first element should be a contract ID. The time resolution
 	// for TTL will be in seconds. The function is executed synchronously and
 	// it returns an error if some error was encountered during storage.
-	PutWithID(contract uint32, messageId, topic, payload []byte, ttl string) error
+	PutWithID(contract uint32, messageId []byte, topic string, payload []byte, ttl string) error
 
 	// Get performs a query and attempts to fetch last messages where
 	// last is specified by last duration argument.
-	Get(contract uint32, topic []byte, last string) ([][]byte, error)
+	Get(contract uint32, topic string, last string) ([][]byte, error)
 
 	// NewID generate messageId that can later used to store and delete message from message store
 	NewID() ([]byte, error)
 
 	// Delete is used to delete entry, the SSID provided must be a full SSID
 	// SSID, where first element should be a contract ID. The function is executed synchronously and
 	// it returns an error if some error was encountered during delete.
-	Delete(contract uint32, messageId, topic []byte) error
+	Delete(contract uint32, messageId []byte, topic string) error
 
 	// PutMessage is used to store a message.
 	// it returns an error if some error was encountered during storage.

server/internal/db/adapter.go

@@ -127,17 +127,17 @@ func (a *adapter) GetName() string {
 }
 
 // Put appends the messages to the store.
-func (a *adapter) Put(contract uint32, topic, payload []byte, ttl string) error {
-	entry := unitdb.NewEntry(topic, payload).WithContract(contract)
+func (a *adapter) Put(contract uint32, topic string, payload []byte, ttl string) error {
+	entry := unitdb.NewEntry([]byte(topic), payload).WithContract(contract)
 	if ttl != "" {
 		entry.WithTTL(ttl)
 	}
 	return a.db.PutEntry(entry)
 }
 
 // PutWithID appends the messages to the store using a pre generated messageId.
-func (a *adapter) PutWithID(contract uint32, messageId, topic, payload []byte, ttl string) error {
-	entry := unitdb.NewEntry(topic, payload).WithContract(contract).WithID(messageId)
+func (a *adapter) PutWithID(contract uint32, messageId []byte, topic string, payload []byte, ttl string) error {
+	entry := unitdb.NewEntry([]byte(topic), payload).WithContract(contract).WithID(messageId)
 	if ttl != "" {
 		entry.WithTTL(ttl)
 	}
@@ -146,9 +146,9 @@ func (a *adapter) PutWithID(contract uint32, messageId, topic, payload []byte, t
 
 // Get performs a query and attempts to fetch last messages where
 // last is specified by last duration argument.
-func (a *adapter) Get(contract uint32, topic []byte, last string) (matches [][]byte, err error) {
+func (a *adapter) Get(contract uint32, topic string, last string) (matches [][]byte, err error) {
 	// Iterating over key/value pairs.
-	query := unitdb.NewQuery(topic).WithContract(contract)
+	query := unitdb.NewQuery([]byte(topic)).WithContract(contract)
 	if last != "" {
 		query.WithLast(last)
 	}
@@ -166,8 +166,8 @@ func (a *adapter) NewID() ([]byte, error) {
 }
 
 // Put appends the messages to the store.
-func (a *adapter) Delete(contract uint32, messageId, topic []byte) error {
-	entry := unitdb.NewEntry(topic, nil)
+func (a *adapter) Delete(contract uint32, messageId []byte, topic string) error {
+	entry := unitdb.NewEntry([]byte(topic), nil)
 	entry.WithContract(contract)
 	return a.db.DeleteEntry(entry.WithID(messageId))
 }