feat(fcm): Migrate topic management to FCM v1 API - #980
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request migrates the topic management functionality in the Firebase Cloud Messaging module from the legacy Instance ID (IID) API to the FCM v1 API, introducing new synchronous and asynchronous methods (subscribe_to_topic, subscribe_to_topic_async, unsubscribe_from_topic, and unsubscribe_from_topic_async) while deprecating the legacy versions. The feedback highlights a potential concurrency race condition in the multi-threaded request execution where a shared headers dictionary is mutated, and suggests extracting duplicated error-handling logic into a single helper method to improve maintainability.
| self._client.request( | ||
| method, | ||
| url=url, | ||
| headers=self._fcm_headers, | ||
| json=json_data, | ||
| ) |
There was a problem hiding this comment.
Passing self._fcm_headers directly to self._client.request inside a ThreadPoolExecutor introduces a concurrency race condition. JsonHttpClient.request mutates the passed headers dictionary by calling .update(METRICS_HEADERS) on it. Since self._fcm_headers is a shared instance variable, concurrent mutations from multiple threads can lead to dictionary corruption or RuntimeError: dictionary changed size during iteration.
To prevent this, pass a copy of the headers dictionary instead.
| self._client.request( | |
| method, | |
| url=url, | |
| headers=self._fcm_headers, | |
| json=json_data, | |
| ) | |
| self._client.request( | |
| method, | |
| url=url, | |
| headers=dict(self._fcm_headers), | |
| json=json_data, | |
| ) |
| def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe): | ||
| """Constructs a result dict from a requests error.""" | ||
| if error.response is not None: | ||
| if is_subscribe and error.response.status_code == 409: | ||
| return {'success': True} | ||
| error_dict = {} | ||
| try: | ||
| parsed = error.response.json() | ||
| if isinstance(parsed, dict): | ||
| error_dict = parsed | ||
| except ValueError: | ||
| pass | ||
|
|
||
| error_data = error_dict.get('error') | ||
| if is_subscribe and isinstance(error_data, dict) and ( | ||
| error_data.get('status') == 'ALREADY_EXISTS' | ||
| ): | ||
| return {'success': True} | ||
|
|
||
| error_code = self._get_topic_error_code(error_dict, error.response.status_code) | ||
| return {'success': False, 'error': error_code} | ||
|
|
||
| return {'success': False, 'error': 'UNKNOWN_ERROR'} | ||
|
|
||
| def _build_topic_subscription_result_from_httpx_error(self, error, is_subscribe): | ||
| """Constructs a result dict from an httpx error.""" | ||
| if isinstance(error, httpx.HTTPStatusError): | ||
| if is_subscribe and error.response.status_code == 409: | ||
| return {'success': True} | ||
| error_dict = {} | ||
| try: | ||
| parsed = error.response.json() | ||
| if isinstance(parsed, dict): | ||
| error_dict = parsed | ||
| except ValueError: | ||
| pass | ||
|
|
||
| error_data = error_dict.get('error') | ||
| if is_subscribe and isinstance(error_data, dict) and ( | ||
| error_data.get('status') == 'ALREADY_EXISTS' | ||
| ): | ||
| return {'success': True} | ||
|
|
||
| error_code = self._get_topic_error_code(error_dict, error.response.status_code) | ||
| return {'success': False, 'error': error_code} | ||
|
|
||
| return {'success': False, 'error': 'UNKNOWN_ERROR'} | ||
|
|
There was a problem hiding this comment.
The methods _build_topic_subscription_result_from_requests_error and _build_topic_subscription_result_from_httpx_error share almost identical logic for parsing error responses and handling the ALREADY_EXISTS (409) idempotency case.
To improve maintainability and reduce code duplication, we can extract the common parsing and validation logic into a single private helper method that operates on the response object.
def _build_topic_subscription_result(self, response, is_subscribe):
\"\"\"Constructs a result dict from a response object.\"\"\"
if response is not None:
if is_subscribe and response.status_code == 409:
return {'success': True}
error_dict = {}
try:
parsed = response.json()
if isinstance(parsed, dict):
error_dict = parsed
except ValueError:
pass
error_data = error_dict.get('error')
if is_subscribe and isinstance(error_data, dict) and (
error_data.get('status') == 'ALREADY_EXISTS'
):
return {'success': True}
error_code = self._get_topic_error_code(error_dict, response.status_code)
return {'success': False, 'error': error_code}
return {'success': False, 'error': 'UNKNOWN_ERROR'}
def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe):
\"\"\"Constructs a result dict from a requests error.\"\"\"
return self._build_topic_subscription_result(error.response, is_subscribe)
def _build_topic_subscription_result_from_httpx_error(self, error, is_subscribe):
\"\"\"Constructs a result dict from an httpx error.\"\"\"
response = error.response if isinstance(error, httpx.HTTPStatusError) else None
return self._build_topic_subscription_result(response, is_subscribe)1 participant
Migrates
subscribe_to_topicandunsubscribe_from_topicin the messaging module from the legacy Instance ID (IID) API to the FCM v1 Topic Subscriptions API.Key changes:
subscribe_to_topicandunsubscribe_from_topicto call the FCM v1 endpoints (/v1/projects/{projectId}/registrations/{token}/topicSubscriptions).subscribe_to_topic_asyncandunsubscribe_from_topic_asyncutilizingHttpxAsyncClient(HTTP/2).subscribe_to_topic_legacyandunsubscribe_from_topic_legacywith deprecation warnings.ALREADY_EXISTS(HTTP 409) is treated as success for topic subscriptions, whileNOT_FOUND(HTTP 404) is recorded as a failure for topic unsubscriptions.