diff --git a/src/client/client.go b/src/client/client.go index 276eed6..1df7b47 100644 --- a/src/client/client.go +++ b/src/client/client.go @@ -406,10 +406,11 @@ type SignalClient struct { signalCliApiConfig *utils.SignalCliApiConfig cliClient *CliClient receiveWebhookUrl string + receiveManualMode bool } func NewSignalClient(signalCliConfig string, attachmentTmpDir string, avatarTmpDir string, signalCliMode SignalCliMode, - jsonRpc2ClientConfigPath string, signalCliApiConfigPath string, receiveWebhookUrl string) *SignalClient { + jsonRpc2ClientConfigPath string, signalCliApiConfigPath string, receiveWebhookUrl string, receiveManualMode bool) *SignalClient { return &SignalClient{ signalCliConfig: signalCliConfig, attachmentTmpDir: attachmentTmpDir, @@ -419,6 +420,7 @@ func NewSignalClient(signalCliConfig string, attachmentTmpDir string, avatarTmpD jsonRpc2Clients: make(map[string]*JsonRpc2Client), signalCliApiConfigPath: signalCliApiConfigPath, receiveWebhookUrl: receiveWebhookUrl, + receiveManualMode: receiveManualMode, } } @@ -442,7 +444,7 @@ func (s *SignalClient) Init(maxRetries int) error { tcpPortsNumberMapping := s.jsonRpc2ClientConfig.GetTcpPortsForNumbers() for number, tcpPort := range tcpPortsNumberMapping { - s.jsonRpc2Clients[number] = NewJsonRpc2Client(s.signalCliApiConfig, number) + s.jsonRpc2Clients[number] = NewJsonRpc2Client(s.signalCliApiConfig, number, s.receiveManualMode) err := s.jsonRpc2Clients[number].Dial("127.0.0.1:"+strconv.FormatInt(tcpPort, 10), maxRetries) if err != nil { return err diff --git a/src/client/jsonrpc2.go b/src/client/jsonrpc2.go index 93f171a..69935eb 100644 --- a/src/client/jsonrpc2.go +++ b/src/client/jsonrpc2.go @@ -66,23 +66,25 @@ type receiveSubscription struct { } type JsonRpc2Client struct { - conn net.Conn - receivedResponsesById map[string]chan JsonRpc2MessageResponse - receivedMessagesChannels map[string]chan JsonRpc2ReceivedMessage - receiveSubscriptions map[string]*receiveSubscription // account -> sub state - channelAccountByUuid map[string]string // channelUuid -> account - signalCliApiConfig *utils.SignalCliApiConfig - number string - receivedMessagesMutex sync.Mutex - receivedResponsesMutex sync.Mutex - receiveSubscriptionsMutex sync.Mutex - address string + conn net.Conn + receivedResponsesById map[string]chan JsonRpc2MessageResponse + receivedMessagesChannels map[string]chan JsonRpc2ReceivedMessage + receiveSubscriptions map[string]*receiveSubscription // account -> sub state + channelAccountByUuid map[string]string // channelUuid -> account + signalCliApiConfig *utils.SignalCliApiConfig + number string + manualReceiveMode bool + receivedMessagesMutex sync.Mutex + receivedResponsesMutex sync.Mutex + receiveSubscriptionsMutex sync.Mutex + address string } -func NewJsonRpc2Client(signalCliApiConfig *utils.SignalCliApiConfig, number string) *JsonRpc2Client { +func NewJsonRpc2Client(signalCliApiConfig *utils.SignalCliApiConfig, number string, manualRcpReceiveMode bool) *JsonRpc2Client { return &JsonRpc2Client{ signalCliApiConfig: signalCliApiConfig, number: number, + manualReceiveMode: manualRcpReceiveMode, receivedResponsesById: make(map[string]chan JsonRpc2MessageResponse), receivedMessagesChannels: make(map[string]chan JsonRpc2ReceivedMessage), receiveSubscriptions: make(map[string]*receiveSubscription), @@ -301,11 +303,6 @@ func (r *JsonRpc2Client) ReceiveData(number string, receiveWebhookUrl string) { } } -// subscribeReceive starts receiving messages for an account by calling -// signal-cli's subscribeReceive JSON-RPC method. Only relevant when -// signal-cli was launched with --receive-mode=manual; in auto mode the -// daemon pushes notifications without an explicit subscribe call. -// Returns the subscription id assigned by signal-cli. func (r *JsonRpc2Client) subscribeReceive(account string) (int64, error) { type subscribeReceiveArgs struct{} resultStr, err := r.getRaw("subscribeReceive", &account, subscribeReceiveArgs{}) @@ -319,8 +316,6 @@ func (r *JsonRpc2Client) subscribeReceive(account string) (int64, error) { return subscriptionId, nil } -// unsubscribeReceive cancels a manual-mode subscription previously -// returned by subscribeReceive. func (r *JsonRpc2Client) unsubscribeReceive(account string, subscriptionId int64) error { type unsubscribeReceiveArgs struct { Subscription int64 `json:"subscription"` @@ -329,10 +324,6 @@ func (r *JsonRpc2Client) unsubscribeReceive(account string, subscriptionId int64 return err } -// acquireReceiveSubscription ensures an active manual-mode subscription -// exists for the given account, refcounting concurrent websocket -// subscribers. The first caller for an account triggers a real -// subscribeReceive RPC; subsequent callers just bump the refcount. func (r *JsonRpc2Client) acquireReceiveSubscription(account string) error { r.receiveSubscriptionsMutex.Lock() defer r.receiveSubscriptionsMutex.Unlock() @@ -341,45 +332,42 @@ func (r *JsonRpc2Client) acquireReceiveSubscription(account string) error { sub.refcount++ return nil } + id, err := r.subscribeReceive(account) if err != nil { return err } + r.receiveSubscriptions[account] = &receiveSubscription{id: id, refcount: 1} log.Infof("Subscribed to receive notifications for account %s (subscription=%d)", account, id) return nil } -// releaseReceiveSubscription decrements the per-account refcount and -// cancels the subscription with signal-cli if it drops to zero. -func (r *JsonRpc2Client) releaseReceiveSubscription(account string) { +func (r *JsonRpc2Client) releaseReceiveSubscription(account string) error { r.receiveSubscriptionsMutex.Lock() defer r.receiveSubscriptionsMutex.Unlock() sub, ok := r.receiveSubscriptions[account] if !ok { - return + return nil } + sub.refcount-- if sub.refcount > 0 { - return + return nil } - if err := r.unsubscribeReceive(account, sub.id); err != nil { + + err := r.unsubscribeReceive(account, sub.id) + if err != nil { log.Warnf("unsubscribeReceive failed for account %s (subscription=%d): %s", account, sub.id, err.Error()) } else { log.Infof("Unsubscribed from receive notifications for account %s (subscription=%d)", account, sub.id) } + delete(r.receiveSubscriptions, account) + return err } -// GetReceiveChannel returns a channel that will receive messages for the -// given account. If signal-cli is in manual receive-mode, it also acquires -// a subscription so that signal-cli starts forwarding messages for the -// account; in auto mode, account is unused (notifications flow regardless). -// -// account may be empty when the caller does not need a subscription -// (e.g. legacy callers in auto mode); in that case no subscribeReceive -// RPC is issued. func (r *JsonRpc2Client) GetReceiveChannel(account string) (chan JsonRpc2ReceivedMessage, string, error) { c := make(chan JsonRpc2ReceivedMessage, 64) @@ -388,8 +376,10 @@ func (r *JsonRpc2Client) GetReceiveChannel(account string) (chan JsonRpc2Receive return c, "", err } - if err := r.acquireReceiveSubscription(account); err != nil { - return c, "", fmt.Errorf("subscribeReceive failed for account %s: %w", account, err) + if r.manualReceiveMode { + if err := r.acquireReceiveSubscription(account); err != nil { + return c, "", fmt.Errorf("Couldn't create subscription for account %s: %w", account, err) + } } r.receivedMessagesMutex.Lock() @@ -407,7 +397,9 @@ func (r *JsonRpc2Client) RemoveReceiveChannel(channelUuid string) { delete(r.channelAccountByUuid, channelUuid) r.receivedMessagesMutex.Unlock() - if account != "" { - r.releaseReceiveSubscription(account) + if r.manualReceiveMode { + if err := r.releaseReceiveSubscription(account); err != nil { + log.Errorf("Couldn't release subscription for account %s: %s", account, err.Error()) + } } } diff --git a/src/main.go b/src/main.go index ede0cc1..bbeb8fa 100644 --- a/src/main.go +++ b/src/main.go @@ -156,6 +156,8 @@ func main() { } } + signalCliManualReceive := utils.GetEnv("JSON_RPC_RECEIVE_MODE", "") == "manual" + webhookUrl := utils.GetEnv("RECEIVE_WEBHOOK_URL", "") if webhookUrl != "" && signalCliMode != client.JsonRpc { log.Fatal("Env variable RECEIVE_WEBHOOK_URL can only be used with mode json-rpc!") @@ -163,7 +165,7 @@ func main() { jsonRpc2ClientConfigPathPath := *signalCliConfig + "/jsonrpc2.yml" signalCliApiConfigPath := *signalCliConfig + "/api-config.yml" - signalClient := client.NewSignalClient(*signalCliConfig, *attachmentTmpDir, *avatarTmpDir, signalCliMode, jsonRpc2ClientConfigPathPath, signalCliApiConfigPath, webhookUrl) + signalClient := client.NewSignalClient(*signalCliConfig, *attachmentTmpDir, *avatarTmpDir, signalCliMode, jsonRpc2ClientConfigPathPath, signalCliApiConfigPath, webhookUrl, signalCliManualReceive) err = signalClient.Init(60) if err != nil { log.Fatal("Couldn't init Signal Client: ", err.Error())