Create Basic Api Key connection

ably/__init__.py

@@ -1,4 +1,5 @@
 from ably.rest.rest import AblyRest
+from ably.realtime.realtime import AblyRealtime
 from ably.rest.auth import Auth
 from ably.rest.push import Push
 from ably.types.capability import Capability

ably/realtime/connection.py

@@ -0,0 +1,89 @@
+import logging
+import asyncio
+import websockets
+import json
+from ably.util.exceptions import AblyAuthException
+from enum import Enum, IntEnum
+
+log = logging.getLogger(__name__)
+
+
+class ConnectionState(Enum):
+    INITIALIZED = 'initialized'
+    CONNECTING = 'connecting'
+    CONNECTED = 'connected'
+    CLOSING = 'closing'
+    CLOSED = 'closed'
+    FAILED = 'failed'
+
+
+class ProtocolMessageAction(IntEnum):
+    CONNECTED = 4
+    ERROR = 9
+
+
+class RealtimeConnection:
+    def __init__(self, realtime):
+        self.options = realtime.options
+        self.__ably = realtime
+        self.__state = ConnectionState.INITIALIZED
+        self.__connected_future = None
+        self.__websocket = None
+
+    async def connect(self):
+        if self.__state == ConnectionState.CONNECTED:
+            return
+
+        if self.__state == ConnectionState.CONNECTING:
+            if self.__connected_future is None:
+                log.fatal('Connection state is CONNECTING but connected_future does not exits')
+                return
+            await self.__connected_future
+        else:
+            self.__state = ConnectionState.CONNECTING
+            self.__connected_future = asyncio.Future()
+            asyncio.create_task(self.connect_impl())
+            await self.__connected_future
+            self.__state = ConnectionState.CONNECTED
+
+    async def close(self):
+        self.__state = ConnectionState.CLOSING
+        if self.__websocket:
+            await self.__websocket.close()
+        else:
+            log.warn('Connection.closed called while connection already closed')
+        self.__state = ConnectionState.CLOSED
+
+    async def connect_impl(self):
+        async with websockets.connect(f'wss://{self.options.realtime_host}?key={self.ably.key}') as websocket:
+            self.__websocket = websocket
+            task = asyncio.create_task(self.ws_read_loop())
+            await task
+
+    async def ws_read_loop(self):
+        while True:
+            raw = await self.__websocket.recv()
+            msg = json.loads(raw)
+            action = msg['action']
+            if action == ProtocolMessageAction.CONNECTED:  # CONNECTED
+                if self.__connected_future:
+                    self.__connected_future.set_result(None)
+                    self.__connected_future = None
+                else:
+                    log.warn('CONNECTED message received but connected_future not set')
+            if action == ProtocolMessageAction.ERROR:  # ERROR
+                error = msg["error"]
+                if error['nonfatal'] is False:
+                    self.__state = ConnectionState.FAILED
+                    if self.__connected_future:
+                        self.__connected_future.set_exception(
+                            AblyAuthException(error["message"], error["statusCode"], error["code"]))
+                        self.__connected_future = None
+
+    @property
+    def ably(self):
+        return self.__ably
+
+    @property
+    def state(self):
+        return self.__state

ably/realtime/realtime.py

@@ -0,0 +1,48 @@
+import logging
+from ably.realtime.connection import RealtimeConnection
+from ably.rest.auth import Auth
+from ably.types.options import Options
+
+
+log = logging.getLogger(__name__)
+
+
+class AblyRealtime:
+    """Ably Realtime Client"""
+
+    def __init__(self, key=None, **kwargs):
+        """Create an AblyRealtime instance.
+
+        :Parameters:
+          **Credentials**
+          - `key`: a valid ably key string
+        """
+
+        if key is not None:
+            options = Options(key=key, **kwargs)
+        else:
+            raise ValueError("Key is missing. Provide an API key.")
+
+        self.__auth = Auth(self, options)
+        self.__options = options
+        self.key = key
+        self.__connection = RealtimeConnection(self)
+
+    async def connect(self):
+        await self.connection.connect()
+
+    async def close(self):
+        await self.connection.close()
+
+    @property
+    def auth(self):
+        return self.__auth
+
+    @property
+    def options(self):
+        return self.__options
+
+    @property
+    def connection(self):
+        """Establish realtime connection"""
+        return self.__connection

ably/types/options.py

@@ -27,6 +27,9 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None,
             from ably import api_version
             idempotent_rest_publishing = api_version >= '1.2'
 
+        if realtime_host is None:
+            realtime_host = Defaults.realtime_host
+
         self.__client_id = client_id
         self.__log_level = log_level
         self.__tls = tls

pyproject.toml

@@ -33,6 +33,7 @@ h2 = "^4.0.0"
 # Optional dependencies
 pycrypto = { version = "^2.6.1", optional = true }
 pycryptodome = { version = "*", optional = true }
+websockets = "^10.3"
 
 [tool.poetry.extras]
 oldcrypto = ["pycrypto"]