diff --git a/libgm/client.go b/libgm/client.go index 460dd10..b4b3524 100644 --- a/libgm/client.go +++ b/libgm/client.go @@ -43,10 +43,17 @@ type EventHandler func(evt any) type Client struct { Logger zerolog.Logger - rpc *RPC evHandler EventHandler sessionHandler *SessionHandler + longPollingConn io.Closer + listenID int + skipCount int + disconnecting bool + + recentUpdates [8][32]byte + recentUpdatesPtr int + conversationsFetchedOnce bool AuthData *AuthData @@ -74,7 +81,6 @@ func NewClient(authData *AuthData, logger zerolog.Logger) *Client { http: &http.Client{}, } sessionHandler.client = cli - cli.rpc = &RPC{client: cli} cli.FetchConfigVersion() return cli } @@ -114,7 +120,7 @@ func (c *Client) Connect() error { return fmt.Errorf("failed to get web encryption key: %w", err) } c.updateWebEncryptionKey(webEncryptionKeyResponse.GetKey()) - go c.rpc.ListenReceiveMessages(true) + go c.ListenReceiveMessages(true) c.sessionHandler.startAckInterval() bugleRes, bugleErr := c.IsBugleDefault() @@ -130,12 +136,13 @@ func (c *Client) Connect() error { } func (c *Client) Disconnect() { - c.rpc.CloseConnection() + c.closeLongPolling() c.http.CloseIdleConnections() } func (c *Client) IsConnected() bool { - return c.rpc != nil + // TODO add better check (longPollingConn is set to nil while the polling reconnects) + return c.longPollingConn != nil } func (c *Client) IsLoggedIn() bool { @@ -143,10 +150,7 @@ func (c *Client) IsLoggedIn() bool { } func (c *Client) Reconnect() error { - c.rpc.CloseConnection() - for c.rpc.conn != nil { - time.Sleep(time.Millisecond * 100) - } + c.closeLongPolling() err := c.Connect() if err != nil { c.Logger.Err(err).Msg("Failed to reconnect") diff --git a/libgm/event_handler.go b/libgm/event_handler.go index 6a01923..0eab351 100644 --- a/libgm/event_handler.go +++ b/libgm/event_handler.go @@ -38,7 +38,7 @@ var responseType = map[gmproto.ActionType]proto.Message{ gmproto.ActionType_UPDATE_CONVERSATION: &gmproto.UpdateConversationResponse{}, } -func (r *RPC) decryptInternalMessage(data *gmproto.IncomingRPCMessage) (*IncomingRPCMessage, error) { +func (c *Client) decryptInternalMessage(data *gmproto.IncomingRPCMessage) (*IncomingRPCMessage, error) { msg := &IncomingRPCMessage{ IncomingRPCMessage: data, } @@ -60,7 +60,7 @@ func (r *RPC) decryptInternalMessage(data *gmproto.IncomingRPCMessage) (*Incomin msg.DecryptedMessage = responseStruct.ProtoReflect().New().Interface() } if msg.Message.EncryptedData != nil { - msg.DecryptedData, err = r.client.AuthData.RequestCrypto.Decrypt(msg.Message.EncryptedData) + msg.DecryptedData, err = c.AuthData.RequestCrypto.Decrypt(msg.Message.EncryptedData) if err != nil { return nil, err } @@ -77,21 +77,21 @@ func (r *RPC) decryptInternalMessage(data *gmproto.IncomingRPCMessage) (*Incomin return msg, nil } -func (r *RPC) deduplicateHash(hash [32]byte) bool { - const recentUpdatesLen = len(r.recentUpdates) - for i := r.recentUpdatesPtr + recentUpdatesLen - 1; i >= r.recentUpdatesPtr; i-- { - if r.recentUpdates[i%recentUpdatesLen] == hash { +func (c *Client) deduplicateHash(hash [32]byte) bool { + const recentUpdatesLen = len(c.recentUpdates) + for i := c.recentUpdatesPtr + recentUpdatesLen - 1; i >= c.recentUpdatesPtr; i-- { + if c.recentUpdates[i%recentUpdatesLen] == hash { return true } } - r.recentUpdates[r.recentUpdatesPtr] = hash - r.recentUpdatesPtr = (r.recentUpdatesPtr + 1) % recentUpdatesLen + c.recentUpdates[c.recentUpdatesPtr] = hash + c.recentUpdatesPtr = (c.recentUpdatesPtr + 1) % recentUpdatesLen return false } -func (r *RPC) logContent(res *IncomingRPCMessage) { - if r.client.Logger.Trace().Enabled() && (res.DecryptedData != nil || res.DecryptedMessage != nil) { - evt := r.client.Logger.Trace() +func (c *Client) logContent(res *IncomingRPCMessage) { + if c.Logger.Trace().Enabled() && (res.DecryptedData != nil || res.DecryptedMessage != nil) { + evt := c.Logger.Trace() if res.DecryptedMessage != nil { evt.Str("proto_name", string(res.DecryptedMessage.ProtoReflect().Descriptor().FullName())) } @@ -104,47 +104,47 @@ func (r *RPC) logContent(res *IncomingRPCMessage) { } } -func (r *RPC) deduplicateUpdate(msg *IncomingRPCMessage) bool { +func (c *Client) deduplicateUpdate(msg *IncomingRPCMessage) bool { if msg.DecryptedData != nil { contentHash := sha256.Sum256(msg.DecryptedData) - if r.deduplicateHash(contentHash) { - r.client.Logger.Trace().Hex("data_hash", contentHash[:]).Msg("Ignoring duplicate update") + if c.deduplicateHash(contentHash) { + c.Logger.Trace().Hex("data_hash", contentHash[:]).Msg("Ignoring duplicate update") return true } - r.logContent(msg) + c.logContent(msg) } return false } -func (r *RPC) HandleRPCMsg(rawMsg *gmproto.IncomingRPCMessage) { - msg, err := r.decryptInternalMessage(rawMsg) +func (c *Client) HandleRPCMsg(rawMsg *gmproto.IncomingRPCMessage) { + msg, err := c.decryptInternalMessage(rawMsg) if err != nil { - r.client.Logger.Err(err).Msg("Failed to decode incoming RPC message") + c.Logger.Err(err).Msg("Failed to decode incoming RPC message") return } - r.client.sessionHandler.queueMessageAck(msg.ResponseID) - if r.client.sessionHandler.receiveResponse(msg) { + c.sessionHandler.queueMessageAck(msg.ResponseID) + if c.sessionHandler.receiveResponse(msg) { return } switch msg.BugleRoute { case gmproto.BugleRoute_PairEvent: - go r.client.handlePairingEvent(msg) + go c.handlePairingEvent(msg) case gmproto.BugleRoute_DataEvent: - if r.skipCount > 0 { - r.skipCount-- - r.client.Logger.Debug(). + if c.skipCount > 0 { + c.skipCount-- + c.Logger.Debug(). Any("action", msg.Message.GetAction()). - Int("remaining_skip_count", r.skipCount). + Int("remaining_skip_count", c.skipCount). Msg("Skipped DataEvent") if msg.DecryptedMessage != nil { - r.client.Logger.Trace(). + c.Logger.Trace(). Str("proto_name", string(msg.DecryptedMessage.ProtoReflect().Descriptor().FullName())). Str("data", base64.StdEncoding.EncodeToString(msg.DecryptedData)). Msg("Skipped event data") } return } - r.client.handleUpdatesEvent(msg) + c.handleUpdatesEvent(msg) } } diff --git a/libgm/pair.go b/libgm/pair.go index ab09a1e..e1ae2e7 100644 --- a/libgm/pair.go +++ b/libgm/pair.go @@ -19,7 +19,7 @@ func (c *Client) StartLogin() (string, error) { return "", err } c.AuthData.TachyonAuthToken = registered.AuthKeyData.TachyonAuthToken - go c.rpc.ListenReceiveMessages(false) + go c.ListenReceiveMessages(false) qr, err := c.GenerateQRCodeData(registered.GetPairingKey()) if err != nil { return "", fmt.Errorf("failed to generate QR code: %w", err) diff --git a/libgm/rpc.go b/libgm/rpc.go index dbc90c4..404b4a1 100644 --- a/libgm/rpc.go +++ b/libgm/rpc.go @@ -20,113 +20,89 @@ import ( "go.mau.fi/mautrix-gmessages/libgm/util" ) -type RPC struct { - client *Client - conn io.ReadCloser - stopping bool - listenID int - - skipCount int - - recentUpdates [8][32]byte - recentUpdatesPtr int -} - -func (r *RPC) ListenReceiveMessages(loggedIn bool) { - r.listenID++ - listenID := r.listenID +func (c *Client) ListenReceiveMessages(loggedIn bool) { + c.listenID++ + listenID := c.listenID errored := true listenReqID := uuid.NewString() - for r.listenID == listenID { - err := r.client.refreshAuthToken() + for c.listenID == listenID { + err := c.refreshAuthToken() if err != nil { - r.client.Logger.Err(err).Msg("Error refreshing auth token") + c.Logger.Err(err).Msg("Error refreshing auth token") if loggedIn { - r.client.triggerEvent(&events.ListenFatalError{Error: fmt.Errorf("failed to refresh auth token: %w", err)}) + c.triggerEvent(&events.ListenFatalError{Error: fmt.Errorf("failed to refresh auth token: %w", err)}) } return } - r.client.Logger.Debug().Msg("Starting new long-polling request") + c.Logger.Debug().Msg("Starting new long-polling request") payload := &gmproto.ReceiveMessagesRequest{ Auth: &gmproto.AuthMessage{ RequestID: listenReqID, - TachyonAuthToken: r.client.AuthData.TachyonAuthToken, + TachyonAuthToken: c.AuthData.TachyonAuthToken, ConfigVersion: util.ConfigMessage, }, Unknown: &gmproto.ReceiveMessagesRequest_UnknownEmptyObject2{ Unknown: &gmproto.ReceiveMessagesRequest_UnknownEmptyObject1{}, }, } - resp, err := r.client.makeProtobufHTTPRequest(util.ReceiveMessagesURL, payload, ContentTypePBLite) + resp, err := c.makeProtobufHTTPRequest(util.ReceiveMessagesURL, payload, ContentTypePBLite) if err != nil { if loggedIn { - r.client.triggerEvent(&events.ListenTemporaryError{Error: err}) + c.triggerEvent(&events.ListenTemporaryError{Error: err}) } errored = true - r.client.Logger.Err(err).Msg("Error making listen request, retrying in 5 seconds") + c.Logger.Err(err).Msg("Error making listen request, retrying in 5 seconds") time.Sleep(5 * time.Second) continue } if resp.StatusCode >= 400 && resp.StatusCode < 500 { - r.client.Logger.Error().Int("status_code", resp.StatusCode).Msg("Error making listen request") + c.Logger.Error().Int("status_code", resp.StatusCode).Msg("Error making listen request") if loggedIn { - r.client.triggerEvent(&events.ListenFatalError{Error: events.HTTPError{Action: "polling", Resp: resp}}) + c.triggerEvent(&events.ListenFatalError{Error: events.HTTPError{Action: "polling", Resp: resp}}) } return } else if resp.StatusCode >= 500 { if loggedIn { - r.client.triggerEvent(&events.ListenTemporaryError{Error: events.HTTPError{Action: "polling", Resp: resp}}) + c.triggerEvent(&events.ListenTemporaryError{Error: events.HTTPError{Action: "polling", Resp: resp}}) } errored = true - r.client.Logger.Debug().Int("statusCode", resp.StatusCode).Msg("5xx error in long polling, retrying in 5 seconds") + c.Logger.Debug().Int("statusCode", resp.StatusCode).Msg("5xx error in long polling, retrying in 5 seconds") time.Sleep(5 * time.Second) continue } if errored { errored = false if loggedIn { - r.client.triggerEvent(&events.ListenRecovered{}) + c.triggerEvent(&events.ListenRecovered{}) } } - r.client.Logger.Debug().Int("statusCode", resp.StatusCode).Msg("Long polling opened") - r.conn = resp.Body - if r.client.AuthData.Browser != nil { + c.Logger.Debug().Int("statusCode", resp.StatusCode).Msg("Long polling opened") + c.longPollingConn = resp.Body + if c.AuthData.Browser != nil { go func() { - err := r.client.NotifyDittoActivity() + err := c.NotifyDittoActivity() if err != nil { - r.client.Logger.Err(err).Msg("Error notifying ditto activity") + c.Logger.Err(err).Msg("Error notifying ditto activity") } }() } - r.startReadingData(resp.Body) - r.conn = nil + c.startReadingData(resp.Body) + c.longPollingConn = nil } } -/* - The start of a message always begins with byte 44 (",") - If the message is parsable (after , has been removed) as an array of interfaces: - func (r *RPC) tryUnmarshalJSON(jsonData []byte, msgArr *[]interface{}) error { - err := json.Unmarshal(jsonData, &msgArr) - return err - } - then the message is complete and it should continue to the HandleRPCMsg function and it should also reset the buffer so that the next message can be received properly. - - if it's not parsable, it should just append the received data to the buf and attempt to parse it until it's parsable. Because that would indicate that the full msg has been received -*/ - -func (r *RPC) startReadingData(rc io.ReadCloser) { - r.stopping = false +func (c *Client) startReadingData(rc io.ReadCloser) { defer rc.Close() + c.disconnecting = false reader := bufio.NewReader(rc) buf := make([]byte, 2621440) var accumulatedData []byte n, err := reader.Read(buf[:2]) if err != nil { - r.client.Logger.Err(err).Msg("Error reading opening bytes") + c.Logger.Err(err).Msg("Error reading opening bytes") return } else if n != 2 || string(buf[:2]) != "[[" { - r.client.Logger.Err(err).Msg("Opening is not [[") + c.Logger.Err(err).Msg("Opening is not [[") return } var expectEOF bool @@ -134,20 +110,20 @@ func (r *RPC) startReadingData(rc io.ReadCloser) { n, err = reader.Read(buf) if err != nil { var logEvt *zerolog.Event - if (errors.Is(err, io.EOF) && expectEOF) || r.stopping { - logEvt = r.client.Logger.Debug() + if (errors.Is(err, io.EOF) && expectEOF) || c.disconnecting { + logEvt = c.Logger.Debug() } else { - logEvt = r.client.Logger.Warn() + logEvt = c.Logger.Warn() } logEvt.Err(err).Msg("Stopped reading data from server") return } else if expectEOF { - r.client.Logger.Warn().Msg("Didn't get EOF after stream end marker") + c.Logger.Warn().Msg("Didn't get EOF after stream end marker") } chunk := buf[:n] if len(accumulatedData) == 0 { if len(chunk) == 2 && string(chunk) == "]]" { - r.client.Logger.Debug().Msg("Got stream end marker") + c.Logger.Debug().Msg("Got stream end marker") expectEOF = true continue } @@ -155,7 +131,7 @@ func (r *RPC) startReadingData(rc io.ReadCloser) { } accumulatedData = append(accumulatedData, chunk...) if !json.Valid(accumulatedData) { - r.client.Logger.Trace().Bytes("data", chunk).Msg("Invalid JSON, reading next chunk") + c.Logger.Trace().Bytes("data", chunk).Msg("Invalid JSON, reading next chunk") continue } currentBlock := accumulatedData @@ -163,33 +139,33 @@ func (r *RPC) startReadingData(rc io.ReadCloser) { msg := &gmproto.LongPollingPayload{} err = pblite.Unmarshal(currentBlock, msg) if err != nil { - r.client.Logger.Err(err).Msg("Error deserializing pblite message") + c.Logger.Err(err).Msg("Error deserializing pblite message") continue } switch { case msg.GetData() != nil: - r.HandleRPCMsg(msg.GetData()) + c.HandleRPCMsg(msg.GetData()) case msg.GetAck() != nil: - r.client.Logger.Debug().Int32("count", msg.GetAck().GetCount()).Msg("Got startup ack count message") - r.skipCount = int(msg.GetAck().GetCount()) + c.Logger.Debug().Int32("count", msg.GetAck().GetCount()).Msg("Got startup ack count message") + c.skipCount = int(msg.GetAck().GetCount()) case msg.GetStartRead() != nil: - r.client.Logger.Trace().Msg("Got startRead message") + c.Logger.Trace().Msg("Got startRead message") case msg.GetHeartbeat() != nil: - r.client.Logger.Trace().Msg("Got heartbeat message") + c.Logger.Trace().Msg("Got heartbeat message") default: - r.client.Logger.Warn(). + c.Logger.Warn(). Str("data", base64.StdEncoding.EncodeToString(currentBlock)). Msg("Got unknown message") } } } -func (r *RPC) CloseConnection() { - if r.conn != nil { - r.client.Logger.Debug().Msg("Closing connection manually") - r.listenID++ - r.stopping = true - r.conn.Close() - r.conn = nil +func (c *Client) closeLongPolling() { + if conn := c.longPollingConn; conn != nil { + c.Logger.Debug().Msg("Closing long polling connection manually") + c.listenID++ + c.disconnecting = true + _ = conn.Close() + c.longPollingConn = nil } } diff --git a/libgm/updates_handler.go b/libgm/updates_handler.go index c372a9d..d26de22 100644 --- a/libgm/updates_handler.go +++ b/libgm/updates_handler.go @@ -16,27 +16,27 @@ func (c *Client) handleUpdatesEvent(msg *IncomingRPCMessage) { switch evt := data.Event.(type) { case *gmproto.UpdateEvents_UserAlertEvent: - c.rpc.logContent(msg) + c.logContent(msg) c.handleUserAlertEvent(msg, evt.UserAlertEvent) case *gmproto.UpdateEvents_SettingsEvent: - c.rpc.logContent(msg) + c.logContent(msg) c.triggerEvent(evt.SettingsEvent) case *gmproto.UpdateEvents_ConversationEvent: - if c.rpc.deduplicateUpdate(msg) { + if c.deduplicateUpdate(msg) { return } c.triggerEvent(evt.ConversationEvent.GetData()) case *gmproto.UpdateEvents_MessageEvent: - if c.rpc.deduplicateUpdate(msg) { + if c.deduplicateUpdate(msg) { return } c.triggerEvent(evt.MessageEvent.GetData()) case *gmproto.UpdateEvents_TypingEvent: - c.rpc.logContent(msg) + c.logContent(msg) c.triggerEvent(evt.TypingEvent.GetData()) default: