implemented receive endpoint via websocket connection

src/api/api.go

@@ -9,6 +9,7 @@ import (
 	"github.com/gabriel-vasile/mimetype"
 	"github.com/gin-gonic/gin"
 	log "github.com/sirupsen/logrus"
+	"github.com/gorilla/websocket"
 
 	"github.com/bbernhard/signal-cli-rest-api/client"
 	utils "github.com/bbernhard/signal-cli-rest-api/utils"
@@ -82,6 +83,12 @@ type SendMessageResponse struct {
 	Timestamp string `json:"timestamp"`
 }
 
+var connectionUpgrader = websocket.Upgrader{
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
+
 type Api struct {
 	signalClient  *client.SignalClient
 }
@@ -273,6 +280,36 @@ func (a *Api) SendV2(c *gin.Context) {
 func (a *Api) Receive(c *gin.Context) {
 	number := c.Param("number")
 
+	if a.signalClient.GetSignalCliMode() == client.JsonRpc {
+		ws, err := connectionUpgrader.Upgrade(c.Writer, c.Request, nil)
+		if err != nil {
+			c.JSON(400, Error{Msg: err.Error()})
+			return
+		}
+		defer ws.Close()
+		for {
+			data, err := a.signalClient.Receive(number, 0)
+			if err == nil {
+				err = ws.WriteMessage(websocket.TextMessage, []byte(data))
+				if err != nil {
+					log.Error("Couldn't write message: " + err.Error())
+					return
+				}
+			} else {
+				errorMsg := Error{Msg: err.Error()}
+				errorMsgBytes, err := json.Marshal(errorMsg)
+				if err != nil {
+					log.Error("Couldn't serialize error message: " + err.Error())
+					return
+				}
+				err = ws.WriteMessage(websocket.TextMessage, errorMsgBytes)
+				if err != nil {
+					log.Error("Couldn't write message: " + err.Error())
+					return
+				}
+			}
+		}
+	} else {
 		timeout := c.DefaultQuery("timeout", "1")
 		timeoutInt, err := strconv.ParseInt(timeout, 10, 32)
 		if err != nil {
@@ -288,6 +325,7 @@ func (a *Api) Receive(c *gin.Context) {
 
 		c.String(200, jsonStr)
 	}
+}
 
 // @Summary Create a new Signal Group.
 // @Tags Groups

src/client/client.go

@@ -297,6 +297,10 @@ func NewSignalClient(signalCliConfig string, attachmentTmpDir string, avatarTmpD
 	}
 }
 
+func (s *SignalClient) GetSignalCliMode() SignalCliMode {
+	return s.signalCliMode
+}
+
 func (s *SignalClient) Init() error {
 	if s.signalCliMode == JsonRpc {
 		s.jsonRpc2ClientConfig = utils.NewJsonRpc2ClientConfig()
@@ -312,6 +316,8 @@ func (s *SignalClient) Init() error {
 			if err != nil {
 				return err
 			}
+
+			go s.jsonRpc2Clients[number].ReceiveData() //receive messages in goroutine
 		}
 	}
 	return nil
@@ -542,8 +548,16 @@ func (s *SignalClient) SendV2(number string, message string, recps []string, bas
 }
 
 func (s *SignalClient) Receive(number string, timeout int64) (string, error) {
-	if s.signalCliMode == Native {
+	if s.signalCliMode == JsonRpc {
-		return "", errors.New(endpointNotSupportedInJsonRpcMode)
+		jsonRpc2Client, err := s.getJsonRpc2Client(number)
+		if err != nil {
+			return "", err
+		}
+		msg := jsonRpc2Client.ReceiveMessage()
+		if msg.Err.Code != 0 {
+			return "", errors.New(msg.Err.Message)
+		}
+		return string(msg.Params), nil
 	} else {
 		command := []string{"--config", s.signalCliConfig, "--output", "json", "-u", number, "receive", "-t", strconv.FormatInt(timeout, 10)}
 

src/client/jsonrpc2.go

@@ -9,8 +9,27 @@ import (
 	"net"
 )
 
+type Error struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+type JsonRpc2MessageResponse struct {
+	Id     string          `json:"id"`
+	Result json.RawMessage `json:"result"`
+	Err    Error           `json:"error"`
+}
+
+type JsonRpc2ReceivedMessage struct {
+	Method string          `json:"method"`
+	Params json.RawMessage `json:"params"`
+	Err    Error           `json:"error"`
+}
+
 type JsonRpc2Client struct {
 	conn                     net.Conn
+	receivedMessageResponses chan JsonRpc2MessageResponse
+	receivedMessages         chan JsonRpc2ReceivedMessage
 }
 
 func NewJsonRpc2Client() *JsonRpc2Client {
@@ -24,6 +43,9 @@ func (r *JsonRpc2Client) Dial(address string) error {
 		return err
 	}
 
+	r.receivedMessageResponses = make(chan JsonRpc2MessageResponse)
+	r.receivedMessages = make(chan JsonRpc2ReceivedMessage)
+
 	return nil
 }
 
@@ -35,17 +57,6 @@ func (r *JsonRpc2Client) getRaw(command string, args interface{}) (string, error
 		Params  interface{} `json:"params,omitempty"`
 	}
 
-	type Error struct {
-		Code    int    `json:"code"`
-		Message string `json:"message"`
-	}
-
-	type Response struct {
-		Id     string          `json:"id"`
-		Result json.RawMessage `json:"result"`
-		Err    Error           `json:"error"`
-	}
-
 	u, err := uuid.NewV4()
 	if err != nil {
 		return "", err
@@ -68,27 +79,56 @@ func (r *JsonRpc2Client) getRaw(command string, args interface{}) (string, error
 		return "", err
 	}
 
-	connbuf := bufio.NewReader(r.conn)
+	var resp JsonRpc2MessageResponse
 	for {
-		str, err := connbuf.ReadString('\n')
+		resp = <-r.receivedMessageResponses
-		if err != nil {
+		if resp.Id == u.String() {
-			return "", err
+			break
+		}
 	}
 
-		var resp Response
-		err = json.Unmarshal([]byte(str), &resp)
-		if err == nil {
-			if resp.Id == u.String() {
-				log.Info("Response1 = ", string(resp.Result))
 	if resp.Err.Code != 0 {
 		return "", errors.New(resp.Err.Message)
 	}
 	return string(resp.Result), nil
 }
+
+func (r *JsonRpc2Client) ReceiveData() {
+	connbuf := bufio.NewReader(r.conn)
+	for {
+		str, err := connbuf.ReadString('\n')
+		if err != nil {
+			log.Error("Couldn't read data: ", err.Error())
+			continue
+		}
+		//log.Info("Received data = ", str)
+
+		var resp1 JsonRpc2ReceivedMessage
+		json.Unmarshal([]byte(str), &resp1)
+		if resp1.Method == "receive" {
+			select {
+			case r.receivedMessages <- resp1:
+				log.Debug("Message sent to golang channel")
+			default:
+				log.Debug("Couldn't send message to golang channel, as there's no receiver")
+			}
+			continue
+		}
+
+		var resp2 JsonRpc2MessageResponse
+		err = json.Unmarshal([]byte(str), &resp2)
+		if err == nil {
+			if resp2.Id != "" {
+				r.receivedMessageResponses <- resp2
+			}
 		} else {
-			log.Info("Response = ", str)
+			log.Error("Received unparsable message: ", str)
+		}
 	}
 }
 
-	return "", errors.New("no data")
+//blocks until message a message is received
+func (r *JsonRpc2Client) ReceiveMessage() JsonRpc2ReceivedMessage {
+	resp := <-r.receivedMessages
+	return resp
 }

src/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/go-openapi/spec v0.19.8 // indirect
 	github.com/go-openapi/swag v0.19.9 // indirect
 	github.com/gofrs/uuid v3.3.0+incompatible
+	github.com/gorilla/websocket v1.4.2
 	github.com/h2non/filetype v1.1.0
 	github.com/mailru/easyjson v0.7.1 // indirect
 	github.com/robfig/cron/v3 v3.0.1

src/go.sum

@@ -69,6 +69,8 @@ github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/protobuf v1.3.3 h1:gyjaxf+svBWX08ZjK86iN9geUJF0H6gp2IRKX6Nf6/I=
 github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/h2non/filetype v1.1.0 h1:Or/gjocJrJRNK/Cri/TDEKFjAR+cfG6eK65NGYB6gBA=
 github.com/h2non/filetype v1.1.0/go.mod h1:319b3zT68BvV+WRj7cwy856M2ehB3HqNOt6sy1HndBY=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=