afonin/PyAPNs2
1
0
Fork 0
Code Issues Wiki Activity

remove hyper from apns2

Browse Source
This commit is contained in:
Ankit Desai
2020-06-23 18:38:47 -04:00
parent e60ca8c4e9
commit f1220b8e0b
3 changed files with 32 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apns2/client.py
+29 -33
View File
@@ -1,4 +1,5 @@
import collections
import httpx
import json
import logging
import time
@@ -8,7 +9,7 @@ from enum import Enum
from threading import Thread
from typing import Dict, Iterable, Optional, Tuple, Union
from .credentials import CertificateCredentials, Credentials
from .credentials import TokenCredentials
from .errors import ConnectionFailed, exception_class_for_reason
# We don't generally need to know about the Credentials subclasses except to
# keep the old API, where APNsClient took a cert_file
@@ -29,7 +30,7 @@ class NotificationType(Enum):
MDM = 'mdm'
RequestStream = collections.namedtuple('RequestStream', ['stream_id', 'token'])
RequestStream = collections.namedtuple('RequestStream', ['token', 'status', 'reason'])
Notification = collections.namedtuple('Notification', ['token', 'payload'])
DEFAULT_APNS_PRIORITY = NotificationPriority.Immediate
@@ -47,15 +48,13 @@ class APNsClient(object):
ALTERNATIVE_PORT = 2197
def __init__(self,
credentials: Union[Credentials, str],
credentials: TokenCredentials,
use_sandbox: bool = False, use_alternative_port: bool = False, proto: Optional[str] = None,
json_encoder: Optional[type] = None, password: Optional[str] = None,
proxy_host: Optional[str] = None, proxy_port: Optional[int] = None,
heartbeat_period: Optional[float] = None) -> None:
if isinstance(credentials, str):
self.__credentials = CertificateCredentials(credentials, password) # type: Credentials
else:
self.__credentials = credentials
self.__credentials = credentials
self._init_connection(use_sandbox, use_alternative_port, proto, proxy_host, proxy_port)
if heartbeat_period:
@@ -67,9 +66,9 @@ class APNsClient(object):
def _init_connection(self, use_sandbox: bool, use_alternative_port: bool, proto: Optional[str],
proxy_host: Optional[str], proxy_port: Optional[int]) -> None:
server = self.SANDBOX_SERVER if use_sandbox else self.LIVE_SERVER
port = self.ALTERNATIVE_PORT if use_alternative_port else self.DEFAULT_PORT
self._connection = self.__credentials.create_connection(server, port, proto, proxy_host, proxy_port)
self.__server = self.SANDBOX_SERVER if use_sandbox else self.LIVE_SERVER
self.__port = self.ALTERNATIVE_PORT if use_alternative_port else self.DEFAULT_PORT
self.__client = httpx.Client(http2=True)
def _start_heartbeat(self, heartbeat_period: float) -> None:
conn_ref = weakref.ref(self._connection)
@@ -142,24 +141,25 @@ class APNsClient(object):
headers['apns-collapse-id'] = collapse_id
url = '/3/device/{}'.format(token_hex)
stream_id = self._connection.request('POST', url, json_payload, headers) # type: int
return stream_id
response = self.__client.post('https://{}{}'.format(self.__server, url), headers=headers, data=json_payload)
return response.status_code, response.text
def get_notification_result(self, stream_id: int) -> Union[str, Tuple[str, str]]:
def get_notification_result(self, status: int, reason: str) -> Union[str, Tuple[str, str]]:
"""
Get result for specified stream
The function returns: 'Success' or 'failure reason' or ('Unregistered', timestamp)
"""
with self._connection.get_response(stream_id) as response:
if response.status == 200:
return 'Success'
else:
raw_data = response.read().decode('utf-8')
data = json.loads(raw_data) # type: Dict[str, str]
if response.status == 410:
return data['reason'], data['timestamp']
else:
return data['reason']
# with self._connection.get_response(stream_id) as response:
if status == 200:
return 'Success'
else:
return reason
# raw_data = response.read().decode('utf-8')
# data = json.loads(raw_data) # type: Dict[str, str]
# if response.status == 410:
# return data['reason'], data['timestamp']
# else:
# return data['reason']
def send_notification_batch(self, notifications: Iterable[Notification], topic: Optional[str] = None,
priority: NotificationPriority = NotificationPriority.Immediate,
@@ -182,7 +182,7 @@ class APNsClient(object):
next_notification = next(notification_iterator, None)
# Make sure we're connected to APNs, so that we receive and process the server's SETTINGS
# frame before starting to send notifications.
self.connect()
# self.connect()
results = {}
open_streams = collections.deque() # type: typing.Deque[RequestStream]
@@ -195,9 +195,9 @@ class APNsClient(object):
self.update_max_concurrent_streams()
if next_notification is not None and len(open_streams) < self.__max_concurrent_streams:
logger.info('Sending to token %s', next_notification.token)
stream_id = self.send_notification_async(next_notification.token, next_notification.payload, topic,
priority, expiration, collapse_id, push_type)
open_streams.append(RequestStream(stream_id, next_notification.token))
status, reason = self.send_notification_async(next_notification.token, next_notification.payload, topic,
priority, expiration, collapse_id, push_type)
open_streams.append(RequestStream(next_notification.token, status, reason))
next_notification = next(notification_iterator, None)
if next_notification is None:
@@ -208,7 +208,7 @@ class APNsClient(object):
# sent new requests or exited the while loop.) Wait for the first outstanding stream
# to return a response.
pending_stream = open_streams.popleft()
result = self.get_notification_result(pending_stream.stream_id)
result = self.get_notification_result(pending_stream.status, pending_stream.reason)
logger.info('Got response for %s: %s', pending_stream.token, result)
results[pending_stream.token] = result
@@ -216,11 +216,7 @@ class APNsClient(object):
def update_max_concurrent_streams(self) -> None:
# Get the max_concurrent_streams setting returned by the server.
# The max_concurrent_streams value is saved in the H2Connection instance that must be
# accessed using a with statement in order to acquire a lock.
# pylint: disable=protected-access
with self._connection._conn as connection:
max_concurrent_streams = connection.remote_settings.max_concurrent_streams
max_concurrent_streams = 100
if max_concurrent_streams == self.__previous_server_max_concurrent_streams:
# The server hasn't issued an updated SETTINGS frame.