Add Observe option for request header of Dots client #10

dots_client/main.go

@@ -178,7 +178,14 @@ func makeServerHandler(env *task.Env) http.HandlerFunc {
 			requestQuerys = tmpPaths[i+1:]
 			break
 		}
-		log.Debugf("Parsed URI, requestName=%+v, requestQuerys=%+v", requestName, requestQuerys)
+		// map observe option
+		observeStr := r.Header.Get(string(messages.OBSERVE))
+		options := make(map[messages.Option]string)
+		if observeStr != "" {
+			options[messages.OBSERVE] = observeStr
+		}
+
+		log.Debugf("Parsed URI, requestName=%+v, requestQuerys=%+v, options=%+v", requestName, requestQuerys, options)
 
 		if requestName == "" || (!isClientConfigRequest(requestName) && !messages.IsRequest(requestName)) {
 			fmt.Printf("dots_client.serverHandler -- %s is invalid request name \n", requestName)
@@ -215,7 +222,7 @@ func makeServerHandler(env *task.Env) http.HandlerFunc {
 			return
 		}
 
-		err := sendRequest(jsonData, requestName, r.Method, requestQuerys, env)
+		err := sendRequest(jsonData, requestName, r.Method, requestQuerys, env, options)
 		if err != nil {
 			fmt.Printf("dots_client.serverHandler -- %s", err.Error())
 			w.WriteHeader(http.StatusInternalServerError)
@@ -239,7 +246,7 @@ func isClientConfigRequest(requestName string) bool {
 /*
  * sendRequest is a function that sends requests to the server.
  */
-func sendRequest(jsonData []byte, requestName, method string, queryParams []string, env *task.Env) (err error) {
+func sendRequest(jsonData []byte, requestName, method string, queryParams []string, env *task.Env, options map[messages.Option]string) (err error) {
 	if jsonData != nil {
 		err = common.ValidateJson(requestName, string(jsonData))
 		if err != nil {
@@ -252,7 +259,7 @@ func sendRequest(jsonData []byte, requestName, method string, queryParams []stri
 	var requestMessage RequestInterface
 	switch messages.GetChannelType(requestName) {
 	case messages.SIGNAL:
-		requestMessage = NewRequest(code, libCoapType, method, requestName, queryParams, env)
+		requestMessage = NewRequest(code, libCoapType, method, requestName, queryParams, env, options)
 	case messages.DATA:
 		errorMsg := fmt.Sprintf("unsupported channel type error: %s", requestName)
 		log.Errorf("dots_client.sendRequest -- %s", errorMsg)

dots_client/request.go

@@ -8,6 +8,7 @@ import (
 	"reflect"
 	"strings"
 	"time"
+	"strconv"
 
 	log "github.com/sirupsen/logrus"
 	"github.com/ugorji/go/codec"
@@ -25,6 +26,13 @@ type RequestInterface interface {
 	Send()
 }
 
+const (
+	Register     uint16 = 0
+	Deregister   uint16 = 1
+)
+
+var Token map[string][]byte = make(map[string][]byte)
+
 /*
  * Dots requests
  */
@@ -38,12 +46,13 @@ type Request struct {
 	queryParams []string
 
 	env         *task.Env
+	options     map[messages.Option]string
 }
 
 /*
  * Request constructor.
  */
-func NewRequest(code messages.Code, coapType libcoap.Type, method string, requestName string, queryParams []string, env *task.Env) *Request {
+func NewRequest(code messages.Code, coapType libcoap.Type, method string, requestName string, queryParams []string, env *task.Env, options map[messages.Option]string) *Request {
 	return &Request{
 		nil,
 		code,
@@ -53,6 +62,7 @@ func NewRequest(code messages.Code, coapType libcoap.Type, method string, reques
 		requestName,
 		queryParams,
 		env,
+		options,
 	}
 }
 
@@ -112,6 +122,7 @@ func (r *Request) pathString() {
  */
 func (r *Request) CreateRequest() {
 	var code libcoap.Code
+	var observe uint16
 
 	switch strings.ToUpper(r.method) {
 	case "GET":
@@ -132,6 +143,26 @@ func (r *Request) CreateRequest() {
 	r.pdu.MessageID = r.env.CoapSession().NewMessageID()
 	r.pdu.Token = dots_common.RandStringBytes(8)
 	r.pdu.Options = make([]libcoap.Option, 0)
+	observeStr := r.options[messages.OBSERVE]
+	if observeStr != "" {
+		observeValue, error := strconv.ParseUint(observeStr, 10, 16)
+		if error != nil {
+			log.Errorf("Observe is not uint type.")
+			return
+		}
+		observe = uint16(observeValue)
+	} 
+
+	if observe == Register || observe == Deregister {
+		r.pdu.Options = append(r.pdu.Options, libcoap.OptionObserve.Uint16(observe))
+		queryString := queryParamsToString(r.queryParams)
+		if observe == Register {
+			Token[queryString] = r.pdu.Token
+		} else {
+			r.pdu.Token = Token[queryString]
+			delete(Token, queryString)
+		}
+	}
 
 	if r.Message != nil {
 		r.pdu.Data = r.dumpCbor()
@@ -268,3 +299,14 @@ func (r *Request) RestartPingTask(pdu *libcoap.Pdu) {
 			pingResponseHandler,
 			pingTimeoutHandler))
 }
+
+func queryParamsToString(queryParams []string) (str string) {
+	str = ""
+	for i, query := range queryParams {
+		if i == 0 {
+			str = query
+		}
+		str += "&" + query
+	}
+	return
+}

dots_common/messages/message.go

@@ -37,7 +37,10 @@ const (
 	REQUEST  Role = iota
 	RESPONSE
 )
-
+type Option string
+const (
+	OBSERVE Option = "Observe"
+)
 type ChannelType int
 
 const (