[Server] add unit transport protocol

Author: unit-adm

README.md

@@ -1,6 +1,6 @@
 # unitdb [![GoDoc](https://godoc.org/github.com/unit-io/unitdb?status.svg)](https://pkg.go.dev/github.com/unit-io/unitdb) [![Go Report Card](https://goreportcard.com/badge/github.com/unit-io/unitdb)](https://goreportcard.com/report/github.com/unit-io/unitdb) [![Build Status](https://travis-ci.org/unit-io/unitdb.svg?branch=master)](https://travis-ci.org/unit-io/unitdb) [![Coverage Status](https://coveralls.io/repos/github/unit-io/unitdb/badge.svg?branch=master)](https://coveralls.io/github/unit-io/unitdb?branch=master)
 
-Unitdb is blazing fast specialized time-series database for microservices, IoT, and realtime internet connected devices. As Unitdb satisfy the requirements for low latency and binary messaging, it is a perfect time-series database for applications such as internet of things and internet connected devices.
+Unitdb is blazing fast specialized time-series database for microservices, IoT, and realtime internet connected devices. As Unitdb satisfy the requirements for low latency and binary messaging, it is a perfect time-series database for applications such as internet of things and internet connected devices. The Unitdb Server uses uTP (unit Transport Protocol) for the Client Server messaging. Read [uTP Specification](https://github.com/unit-io/unitdb/tree/master/docs/utp.md).
 
 ```
 Don't forget to ⭐ this repo if you like Unitdb!

db.go

@@ -51,6 +51,7 @@ func Open(path string, opts ...Options) (*DB, error) {
 	options := &_Options{}
 	WithDefaultOptions().set(options)
 	WithDefaultFlags().set(options)
+	WithDefaultQueryOptions().set(options)
 	for _, opt := range opts {
 		if opt != nil {
 			opt.set(options)

db_test.go

@@ -92,7 +92,7 @@ func TestSimple(t *testing.T) {
 	var ids [][]byte
 
 	entry := NewEntry(topic, nil)
-	entry.WithContract(contract).WithTTL([]byte("1m"))
+	entry.WithContract(contract).WithTTL("1m")
 	for i = 0; i < n; i++ {
 		messageID := db.NewID()
 		entry.WithID(messageID)

docs/utp.md

@@ -0,0 +1,100 @@
+# Unit Transport Protocol (uTP)
+
+## About uTP
+The uTP (unit Tranport Protocol) specification defines the Client Server message transport protocol. It is light weight, open, simple and designed for communication in Machine to Machine (M2M) and Internet connected devices (IoT) contexts. The uTP protocol runs over TCP/IP, WebSocket, GRPC or other network protocols that provide bi-directional connections.
+
+## Message Flow
+An application transport the data by uTP across network, it contains payload data, delivey mode and topic with optional collection of properties.
+
+### 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 in.
+- 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.
+- Processes topic subscription requests from Clients.
+- Route Application Messages that match Client subscriptions.
+- Closes the network connection from the Client.
+
+## Application Message
+The Application Messages are transported between Client and Server in the form of data Packets. A data Packet consist of Fixed Header and a uTP Message Type defined in the specification.
+
+### Fixed Header
+Each uTP data packet contains a Fixed Header as shown below:
+
+|  Name | Type |
+| :--- | :--- |
+| MessageType |	enum |
+| MessageLength | int32 |
+
+### Message Type
+
+| Name | Value | Direction of Flow |
+| :--- | :--- | :--- |
+| RERSERVED | 0 | Forbidden |
+| CONNECT |	1 |	Client to Server |
+| CONNACK |	2 |	Server to Client |
+| PUBLISH |	3 |	Client to Server or Server to Client |
+| PUBNEW | 4 | Server to Client |
+| PUBRECEIVE | 5 |	Client to Server |
+| PUBRECEIPT | 6 | Client to Server or Server to Client |
+| PUBCOMPLETE |	7 | Client to Server or Server to Client |
+| SUBSCRIBE | 8 | Client to Server |
+| SUBACK | 9 | Server to Client |
+| UNSUBSCRIBE |	10	| Client to Server |
+| UNSUBACK | 11 | Server to Client |
+| PINGREQ |	12 | Client to Server |
+| PINGRESP | 13 | Server to Client |
+| DISCONNECT | 14 | Client to Server or Server to Client |
+
+## Delivery Mode
+The uTP delivers Application Messages as per the Delivery Mode defined in the publish and subscribe Message type. The Client and Server both can be a publisher or subcriber of messages.
+Th Delivery Mode in the oubound Application Message to the Client could differ from that of inbound Application Message.
+
+### Express Delivery
+The Express Delivery Mode ensures that the Application Message arrives at the receiver at least once. An Express Delivery Mode PUBLISH Message has a Message Identifier and get the receipt acknowledgement by a PUBRECEIPT Message.
+
+In the Express Delivery Mode, the sender
+- Must send a PUBLISH Message containing a Message Identifier with Delivery Mode Express
+- Must treat a PUBLISH Message as "unacknowledged" until it has received the corresponding PUBCOMPLETE Message from the receiver.
+
+A sender is permitted to send further PUBLISH Messages with different Message Identifier while it is waiting to receive publish receipts.
+
+In the Express Delivery Mode, the receiver
+- Must respond with a PUBCOMPLETE Message containing the Message Identifier from the incoming PUBLISH Message.
+
+### Reliable Delivery
+The Reliable Delivery Mode ensures no duplication of Application Messages delivered to the receivers. There is an increased overhead associated to the Reliable Delivery Mode.
+
+### Inbound Message Flow from a Client
+In the Reliable Delivery Mode, the sender
+- Must send a PUBLISH Message containing a Message Identifier with Delivery Mode Reliable
+- Must treat a PUBLISH Message as "unacknowledged" until it has received the corresponding PUBRECEIPT Message from the receiver.
+- Must send a PUBCOMPLETE Message when it receive a PUBRECEIPT Message from the receiver. This PUBCOMPLETE Message must contain the same Message Identifier from original PUBLISH Message.
+- Must NOT re-send the PUBLISH Message once it has received the corresponding PUBRECEIPT Message.
+
+A sender is permitted to send further PUBLISH Messages with different Message Identifier while it is waiting to receive publish receipts.
+
+In the Reliable Delivery Mode, the receiver
+- Must respond with a PUBRECEIPT containing the Message Identifier from the incoming PUBLISH Message.
+- Until it has received a corresponding PUBCOMPLETE Message, the receiver MUST acknowledge any subsequent PUBLISH Message with the same Message Identifier by sending a PUBRECEIPT Message.
+
+### Outbound Message Flow to the Client
+In the Reliable Delivery Mode, the sender
+- Must send a PUBNEW Message containing a Message Identifier
+- Must send PUBLISH Message when it receive a PUBRECEIVE Packet from the receiver. This PUBLISH Message must conrain the same Message Identifier from PUBNEW Message.
+- Must treat a PUBLISH Message as "unacknowledged" until it has received the corresponding PUBRECEIPT Message from the receiver.
+- Must send a PUBCOMPLETE Message when it receive a PUBRECEIPT Message from the receiver. This PUBCOMPLETE Message must contain the same Message Identifier from original PUBNEW Message.
+- Must NOT re-send the PUBLISH Message once it has received the corresponding PUBRECEIPT Message.
+
+A sender is permitted to send further PUBLISH Messages with different Message Identifier while it is waiting to receive publish receipts.
+
+In the Reliable Delivery Mode, the receiver
+- Must respond with a PUBRECEIVE containing the Message Identifier from the incoming PUBNEW Message.
+- Must respond with a PUBRECEIPT Message when it receive a PUBLISH Message from the sender containing the Message Identifier from the incoming PUBNEW Message.
+- Until it has received a corresponding PUBCOMPLETE Message, the receiver MUST acknowledge any subsequent PUBLISH Message with the same Message Identifier by sending a PUBRECEIPT Message.
+

entry.go

@@ -77,13 +77,13 @@ func (e *Entry) WithContract(contract uint32) *Entry {
 }
 
 // WithTTL sets TTL for message expiry for the entry.
-func (e *Entry) WithTTL(ttl []byte) *Entry {
-	val, err := strconv.ParseInt(unsafeToString(ttl), 10, 64)
+func (e *Entry) WithTTL(ttl string) *Entry {
+	val, err := strconv.ParseInt(ttl, 10, 64)
 	if err == nil {
 		e.ExpiresAt = uint32(time.Now().Add(time.Duration(int(val)) * time.Second).Unix())
 	}
 	var duration time.Duration
-	duration, _ = time.ParseDuration(unsafeToString(ttl))
+	duration, _ = time.ParseDuration(ttl)
 	e.ExpiresAt = uint32(time.Now().Add(duration).Unix())
 	return e
 }

examples/pubsub/main.go

@@ -16,7 +16,6 @@ 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",
@@ -36,23 +35,23 @@ func main() {
 		log.Fatalf("err: %s", err)
 	}
 
-	r := client.Subscribe("teams.alpha.user1")
+	r := client.Subscribe([]byte("teams.alpha.user1"), unitdb.WithLast("1h"))
 	if _, err := r.Get(ctx, 1*time.Second); err != nil {
 		fmt.Println(err)
 		os.Exit(1)
 	}
 
 	for i := 0; i < 5; i++ {
 		msg := fmt.Sprintf("Hi #%d time!", i)
-		r := client.Publish("teams.alpha.user1", msg)
+		r := client.Publish([]byte("teams.alpha.user1"), []byte(msg), unitdb.WithTTL("1m"))
 		if _, err := r.Get(ctx, 1*time.Second); err != nil {
 			log.Fatalf("err: %s", err)
 		}
 	}
 
 	wait := time.NewTicker(1 * time.Second)
 	<-wait.C
-	r = client.Unsubscribe("teams.alpha.user1")
+	r = client.Unsubscribe([]byte("teams.alpha.user1"))
 	if _, err := r.Get(ctx, 1*time.Second); err != nil {
 		fmt.Println(err)
 		os.Exit(1)

options.go

@@ -163,6 +163,16 @@ func WithBatchWriteInterval(dur time.Duration) Options {
 	})
 }
 
+// WithDefaultQueryOptions will set some default values for Query operation.
+//   defaultQueryLimit: 1000
+//   maxQueryLimit: 100000
+func WithDefaultQueryOptions() Options {
+	return newFuncOption(func(o *_Options) {
+		o.queryOptions.defaultQueryLimit = 1000
+		o.queryOptions.maxQueryLimit = 100000
+	})
+}
+
 // WithDefaultOptions will open DB with some default values.
 func WithDefaultOptions() Options {
 	return newFuncOption(func(o *_Options) {
@@ -172,12 +182,6 @@ func WithDefaultOptions() Options {
 		if o.syncDurationType == 0 {
 			o.syncDurationType = time.Second
 		}
-		if o.queryOptions.defaultQueryLimit == 0 {
-			o.queryOptions.defaultQueryLimit = 1000
-		}
-		if o.queryOptions.maxQueryLimit == 0 {
-			o.queryOptions.maxQueryLimit = 100000
-		}
 		if o.bufferSize == 0 {
 			o.bufferSize = 1 << 32 // maximum size of a buffer to use in bufferpool (4GB).
 		}

query.go

@@ -17,6 +17,8 @@
 package unitdb
 
 import (
+	"time"
+
 	"github.com/unit-io/unitdb/message"
 )
 
@@ -46,8 +48,11 @@ type (
 
 // NewQuery creates a new query structure from the topic.
 func NewQuery(topic []byte) *Query {
+	opts := &_Options{}
+	WithDefaultQueryOptions().set(opts)
 	return &Query{
-		Topic: topic,
+		internal: _InternalQuery{opts: &opts.queryOptions},
+		Topic:    topic,
 	}
 }
 
@@ -63,6 +68,25 @@ func (q *Query) WithLimit(limit int) *Query {
 	return q
 }
 
+// WithLast sets query duration to fetch stored messages.
+func (q *Query) WithLast(dur string) *Query {
+	base := time.Now()
+	duration, err := time.ParseDuration(dur)
+	if err != nil {
+		return q
+	}
+	// In case of last, include it to the query.
+	q.internal.cutoff = base.Add(-duration).Unix()
+	switch {
+	case (q.Limit == 0):
+		q.Limit = q.internal.opts.defaultQueryLimit
+	case q.Limit > q.internal.opts.maxQueryLimit:
+		q.Limit = q.internal.opts.maxQueryLimit
+	}
+
+	return q
+}
+
 func (q *Query) parse() error {
 	if q.Contract == 0 {
 		q.Contract = message.MasterContract

server/internal/cluster.go

@@ -317,11 +317,11 @@ func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
 
 		switch msg.Type {
 		case message.SUBSCRIBE:
-			conn.handle(msg.MsgSub)
+			conn.handler(msg.MsgSub)
 		case message.UNSUBSCRIBE:
-			conn.handle(msg.MsgUnsub)
+			conn.handler(msg.MsgUnsub)
 		case message.PUBLISH:
-			conn.handle(msg.MsgPub)
+			conn.handler(msg.MsgPub)
 		}
 	} else {
 		// Reject the request: wrong signature, cluster is out of sync.