fixed bug in golang channel handling (json-rpc mode)

* golang channels are meant to be 1:1 channels, so if multiple
  goroutines listen on the same channel for messages, only one will
  receive the message and the others are not, which lead to lost
  messages.

  In order to fix that, we create a dedicated golang channel for every
  websocket connection.

see #451
This commit is contained in:
Bernhard B
2023-12-11 22:18:23 +01:00
parent 844f1d7b91
commit 3d7b73560a
3 changed files with 38 additions and 15 deletions

View File

@@ -390,7 +390,7 @@ func (a *Api) SendV2(c *gin.Context) {
}
func (a *Api) handleSignalReceive(ws *websocket.Conn, number string, stop chan struct{}) {
receiveChannel, err := a.signalClient.GetReceiveChannel()
receiveChannel, channelUuid, err := a.signalClient.GetReceiveChannel()
if err != nil {
log.Error("Couldn't get receive channel: ", err.Error())
return
@@ -399,6 +399,7 @@ func (a *Api) handleSignalReceive(ws *websocket.Conn, number string, stop chan s
for {
select {
case <-stop:
a.signalClient.RemoveReceiveChannel(channelUuid)
ws.Close()
return
case msg := <-receiveChannel:

View File

@@ -730,12 +730,20 @@ func (s *SignalClient) Receive(number string, timeout int64, ignoreAttachments b
}
}
func (s *SignalClient) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, error) {
func (s *SignalClient) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, string, error) {
jsonRpc2Client, err := s.getJsonRpc2Client()
if err != nil {
return nil, err
return nil, "", err
}
return jsonRpc2Client.GetReceiveChannel(), nil
return jsonRpc2Client.GetReceiveChannel()
}
func (s *SignalClient) RemoveReceiveChannel(channelUuid string) {
jsonRpc2Client, err := s.getJsonRpc2Client()
if err != nil {
return
}
jsonRpc2Client.RemoveReceiveChannel(channelUuid)
}
func (s *SignalClient) CreateGroup(number string, name string, members []string, description string, editGroupPermission GroupPermission, addMembersPermission GroupPermission, groupLinkState GroupLinkState) (string, error) {

View File

@@ -33,7 +33,7 @@ type JsonRpc2ReceivedMessage struct {
type JsonRpc2Client struct {
conn net.Conn
receivedResponsesById map[string]chan JsonRpc2MessageResponse
receivedMessages chan JsonRpc2ReceivedMessage
receivedMessagesChannels map[string]chan JsonRpc2ReceivedMessage
lastTimeErrorMessageSent time.Time
signalCliApiConfig *utils.SignalCliApiConfig
number string
@@ -44,6 +44,7 @@ func NewJsonRpc2Client(signalCliApiConfig *utils.SignalCliApiConfig, number stri
signalCliApiConfig: signalCliApiConfig,
number: number,
receivedResponsesById: make(map[string]chan JsonRpc2MessageResponse),
receivedMessagesChannels: make(map[string]chan JsonRpc2ReceivedMessage),
}
}
@@ -54,8 +55,6 @@ func (r *JsonRpc2Client) Dial(address string) error {
return err
}
r.receivedMessages = make(chan JsonRpc2ReceivedMessage)
return nil
}
@@ -143,13 +142,15 @@ func (r *JsonRpc2Client) ReceiveData(number string) {
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")
for _, c := range r.receivedMessagesChannels {
select {
case c <- resp1:
log.Debug("Message sent to golang channel")
default:
log.Debug("Couldn't send message to golang channel, as there's no receiver")
}
continue
}
continue
}
var resp2 JsonRpc2MessageResponse
@@ -166,6 +167,19 @@ func (r *JsonRpc2Client) ReceiveData(number string) {
}
}
func (r *JsonRpc2Client) GetReceiveChannel() chan JsonRpc2ReceivedMessage {
return r.receivedMessages
func (r *JsonRpc2Client) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, string, error) {
c := make(chan JsonRpc2ReceivedMessage)
channelUuid, err := uuid.NewV4()
if err != nil {
return c, "", err
}
r.receivedMessagesChannels[channelUuid.String()] = c
return c, channelUuid.String(), nil
}
func (r *JsonRpc2Client) RemoveReceiveChannel(channelUuid string) {
delete(r.receivedMessagesChannels, channelUuid)
}