Merge pull request #1 from pocketlabs/remove_hyper
remove hyper from apns2
This commit is contained in:
Ankit Desai
+25
-29
@@ -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,14 +48,12 @@ 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._init_connection(use_sandbox, use_alternative_port, proto, proxy_host, proxy_port)
|
||||
|
||||
@@ -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:
|
||||
# with self._connection.get_response(stream_id) as response:
|
||||
if 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']
|
||||
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,
|
||||
status, reason = 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))
|
||||
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.
|
||||
|
||||
+1
-25
@@ -3,43 +3,19 @@ from typing import Optional, Tuple, TYPE_CHECKING
|
||||
|
||||
import jwt
|
||||
|
||||
from hyper import HTTP20Connection # type: ignore
|
||||
from hyper.tls import init_context # type: ignore
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from hyper.ssl_compat import SSLContext # type: ignore
|
||||
|
||||
DEFAULT_TOKEN_LIFETIME = 3600
|
||||
DEFAULT_TOKEN_ENCRYPTION_ALGORITHM = 'ES256'
|
||||
|
||||
|
||||
# Abstract Base class. This should not be instantiated directly.
|
||||
class Credentials(object):
|
||||
def __init__(self, ssl_context: 'Optional[SSLContext]' = None) -> None:
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.__ssl_context = ssl_context
|
||||
|
||||
# Creates a connection with the credentials, if available or necessary.
|
||||
def create_connection(self, server: str, port: int, proto: Optional[str], proxy_host: Optional[str] = None,
|
||||
proxy_port: Optional[int] = None) -> HTTP20Connection:
|
||||
# self.__ssl_context may be none, and that's fine.
|
||||
return HTTP20Connection(server, port, ssl_context=self.__ssl_context, force_proto=proto or 'h2',
|
||||
secure=True, proxy_host=proxy_host, proxy_port=proxy_port)
|
||||
|
||||
def get_authorization_header(self, topic: Optional[str]) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
# Credentials subclass for certificate authentication
|
||||
class CertificateCredentials(Credentials):
|
||||
def __init__(self, cert_file: Optional[str] = None, password: Optional[str] = None,
|
||||
cert_chain: Optional[str] = None) -> None:
|
||||
ssl_context = init_context(cert=cert_file, cert_password=password)
|
||||
if cert_chain:
|
||||
ssl_context.load_cert_chain(cert_chain)
|
||||
super(CertificateCredentials, self).__init__(ssl_context)
|
||||
|
||||
|
||||
# Credentials subclass for JWT token based authentication
|
||||
class TokenCredentials(Credentials):
|
||||
def __init__(self, auth_key_path: str, auth_key_id: str, team_id: str,
|
||||
|
||||
Reference in New Issue
Block a user