fix: various fixes of port errors

Author: ivelin

run-dev.sh

@@ -5,5 +5,5 @@ SCRIPTPATH=$(dirname "$SCRIPT")
 echo $SCRIPTPATH
 
 set -ex
-pip3 install --editable src
+# pip3 install --editable src
 python3 $SCRIPTPATH/cli.py

src/peerjs/dataconnection.py

@@ -1,5 +1,5 @@
 """Convenience wrapper around RTCDataChannel."""
-from util import util
+from .util import util
 import logging
 from negotiator import Negotiator
 from enums import \
@@ -12,38 +12,45 @@
 from .encodingqueue import EncodingQueue
 import json
 from aiortc import RTCDataChannel
+import asyncio
+from .peer import Peer
 
 log = logging.getLogger(__name__)
 
 
 class DataConnection(BaseConnection):
- """Wraps a DataChannel between two Peers."""
+ """Wrap a DataChannel between two Peers."""
+
  ID_PREFIX = "dc_"
  MAX_BUFFERED_AMOUNT = 8 * 1024 * 1024
 
  @property
  def type():
+ """Return ConnectionType.Data."""
  return ConnectionType.Data
 
  @property
- def dataChannel() -> RTCDataChannel:
+ def dataChannel(self) -> RTCDataChannel:
+ """Return the RTCDataChannel of this data connection."""
  return self._dc
 
- def bufferSize() -> int:
+ def bufferSize(self) -> int:
+ """Return current data buffer size."""
  return self._bufferSize
 
- def __init__(self, peerId: str, provider: Peer, **options):
+ def __init__(self, peerId: str = None, provider: Peer = None, **options):
+ """Create a DataConnection instance."""
  super(peerId, provider, options)
- _negotiator: Negotiator
- label: str
- serialization: SerializationType
- reliable: boolean
- stringify: lambda data: json.dumps(data)
- parse: lambda jsn: json.loads(jsn)
- _buffer: []
- _bufferSize = 0
- _buffering = False
- _chunkedData = {}
+ self._negotiator: Negotiator = None
+ self.label: str = None
+ self.serialization: SerializationType = None
+ self.reliable: bool = None
+ self.stringify = lambda data: json.dumps(data)
+ self.parse = lambda jsn: json.loads(jsn)
+ self._buffer: []
+ self._bufferSize = 0
+ self._buffering = False
+ self._chunkedData = {}
  # {
  # [id: number]: {
  # data: Blob[],
@@ -52,94 +59,99 @@ def __init__(self, peerId: str, provider: Peer, **options):
  # }
  # }
 
- _dc: RTCDataChannel
- _encodingQueue = EncodingQueue()
- if self.options.connectionId:
- self.connectionId = self.options.connectionId
+ self._dc: RTCDataChannel
+ self._encodingQueue = EncodingQueue()
+ if options.connectionId:
+ self.connectionId = options.connectionId
  else:
  self.connectionId = DataConnection.ID_PREFIX + util.randomToken()
- if self.options.label:
- self.label = this.options.label
+ if options.label:
+ self.label = options.label
  else:
  self.label = self.connectionId
- self.serialization = self.options.serialization if \
- self.options.serialization else SerializationType.Binary
+ self.serialization = options.serialization if \
+ options.serialization else SerializationType.Binary
  self.reliable = self.options.reliable
  @self._encodingQueue.on('done')
- def on_eq_done(ab: ArrayBuffer):
+ def on_eq_done(ab): # ab : ArrayBuffer
  self._bufferedSend(ab)
 
  @self._encodingQueue.on('error')
  def on_eq_error():
- log.error(f'DC#${this.connectionId}: '
+ log.error(f'DC#${self.connectionId}: '
  'Error occured in encoding from blob to arraybuffer, '
  'closing Data Connection.')
  self.close()
  self._negotiator = Negotiator(self)
- payload_option = self.options._payload or {'originator': true}
- self._negotiator.startConnection()
+ payload_option = options._payload or {'originator': True}
+ self._negotiator.startConnection(payload_option)
 
  def initialize(self, dc: RTCDataChannel) -> None:
- """Called by the Negotiator when the DataChannel is ready."""
+ """Configure datachannel when available.
+
+ Called by the Negotiator when the DataChannel is ready.
+ """
  self._dc = dc
  self._configureDataChannel()
 
  def _configureDataChannel(self) -> None:
  if not util.supports.binaryBlob or util.supports.reliable:
- this.dataChannel.binaryType = "arraybuffer"
+ self.dataChannel.binaryType = "arraybuffer"
 
  @self.dataChannel.on('open')
- def on_datachannel_open():
- log.debug(f'DC#${this.connectionId} dc connection success')
+ async def on_datachannel_open():
+ log.debug(f'DC#${self.connectionId} dc connection success')
  self._open = True
  self.emit(ConnectionEventType.Open)
 
  @self.dataChannel.on('message')
- def on_datachannel_message(e):
+ async def on_datachannel_message(e):
  log.debug(f'DC#${self.connectionId} dc onmessage: {e.data}')
- self._handleDataMessage(e)
+ await self._handleDataMessage(e)
 
  @self.dataChannel.on('close')
- def on_datachannel_close(e):
+ async def on_datachannel_close(e):
  log.debug(f'DC#${self.connectionId} dc closed for: {self.peer}')
  self.close()
 
- def _handleDataMessage(self, data) -> None:
- """Handles a DataChannel message."""
- datatype = data.constructor
-
- isBinarySerialization = \
- self.serialization == SerializationType.Binary or \
- self.serialization == SerializationType.BinaryUTF8
-
+ async def _handleDataMessage(self, data) -> None:
+ """Handle a DataChannel message."""
+ # datatype = data.constructor
+ # isBinarySerialization = \
+ # self.serialization == SerializationType.Binary or \
+ # self.serialization == SerializationType.BinaryUTF8
  deserializedData = data
-
- if isBinarySerialization:
- if datatype == Blob:
- # Datatype should never be blob
- ab = await util.blobToArrayBuffer(data)
- unpackedData = util.unpack(ab)
- self.emit(ConnectionEventType.Data, unpackedData)
- return
- if datatype == ArrayBuffer:
- deserializedData = util.unpack(data)
- elif datatype == String:
- # String fallback for binary data for browsers
- # that don't support binary yet
- ab = util.binaryStringToArrayBuffer(data)
- deserializedData = util.unpack(ab)
+ # Peerjs JavaScript version uses a messagepack library
+ # which is not ported to Python yet
+ # if isBinarySerialization:
+ # if datatype == Blob:
+ # # Datatype should never be blob
+ # ab = await util.blobToArrayBuffer(data)
+ # unpackedData = util.unpack(ab)
+ # self.emit(ConnectionEventType.Data, unpackedData)
+ # return
+ # if datatype == ArrayBuffer:
+ # deserializedData = util.unpack(data)
+ # elif datatype == String:
+ # # String fallback for binary data for browsers
+ # # that don't support binary yet
+ # ab = util.binaryStringToArrayBuffer(data)
+ # deserializedData = util.unpack(ab)
+ if self.serialization == SerializationType.Raw:
+ # no special massaging of deserialized data
+ pass
  elif self.serialization == SerializationType.JSON:
  deserializedData = self.parse(data)
 
  # Check if we've chunked--if so, piece things back together.
  # We're guaranteed that this isn't 0.
  if deserializedData.__peerData:
- self._handleChunk(deserializedData)
+ await self._handleChunk(deserializedData)
  return
 
  self.emit(ConnectionEventType.Data, deserializedData)
 
- def _handleChunk(self, data) -> None:
+ async def _handleChunk(self, data) -> None:
  id = data.__peerData
  chunkInfo = self._chunkedData[id] or {
  'data': [],
@@ -148,15 +160,17 @@ def _handleChunk(self, data) -> None:
  }
  chunkInfo.data[data.n] = data.data
  chunkInfo.count += 1
- this._chunkedData[id] = chunkInfo
+ self._chunkedData[id] = chunkInfo
  if chunkInfo.total == chunkInfo.count:
  # Clean up before making
  # the recursive call to `_handleDataMessage`.
  del self._chunkedData[id]
  # We've received all the chunks--time
  # to construct the complete data.
- data = Blob(chunkInfo.data)
- this._handleDataMessage(data)
+ # Blog is a browser JavaScript type.
+ # Not applicable in the Python port.
+ # data = Blob(chunkInfo.data)
+ await self._handleDataMessage(data)
 
  #
  # Exposed functionality for users.
@@ -190,7 +204,7 @@ def send(self, data, chunked: bool) -> None:
  if not self.open:
  self.emit(
  ConnectionEventType.Error,
- RuntimrError(
+ RuntimeError(
  'Connection is not open. '
  'You should listen for the `open` '
  'event before sending messages.'
@@ -199,76 +213,78 @@ def send(self, data, chunked: bool) -> None:
  return
 
  if self.serialization == SerializationType.JSON:
- self._bufferedSend(this.stringify(data))
- elif \
- self.serialization == SerializationType.Binary or \
- self.serialization == SerializationType.BinaryUTF8:
- blob = util.pack(data)
- if not chunked and blob.size > util.chunkedMTU:
- self._sendChunks(blob)
- return
-
- if not util.supports.binaryBlob:
- # We only do this if we really need to
- # (e.g. blobs are not supported),
- # because this conversion is costly.
- self._encodingQueue.enque(blob)
- else:
- self._bufferedSend(blob)
+ self._bufferedSend(self.stringify(data))
+ # Blob is a JavaScript browser type. Not supported in Python.
+ # elif \
+ # self.serialization == SerializationType.Binary or \
+ # self.serialization == SerializationType.BinaryUTF8:
+ # blob = util.pack(data)
+ # if not chunked and blob.size > util.chunkedMTU:
+ # self._sendChunks(blob)
+ # return
+ #
+ # if not util.supports.binaryBlob:
+ # # We only do this if we really need to
+ # # (e.g. blobs are not supported),
+ # # because this conversion is costly.
+ # self._encodingQueue.enque(blob)
+ # else:
+ # self._bufferedSend(blob)
  else:
  self._bufferedSend(data)
 
- def _bufferedSend(self, msg: any) -> None:
- if self._buffering or not self._trySend(msg):
+ async def _bufferedSend(self, msg: any) -> None:
+ if self._buffering or not await self._trySend(msg):
  self._buffer.push(msg)
- self._bufferSize = this._buffer.length
+ self._bufferSize = self._buffer.length
 
  async def _trySend(self, msg) -> bool:
- """Returns true if the send succeeds."""
- if not this.open:
- return false
+ """Return true if the send succeeds."""
+ if not self.open:
+ return False
  if self.dataChannel.bufferedAmount > \
  DataConnection.MAX_BUFFERED_AMOUNT:
- self._buffering = true
+ self._buffering = True
 
- def delayBuf():
+ async def delayBuf():
  # wait for 50ms before trying buffer
  await asyncio.sleep(0.05)
  self._buffering = False
  self._tryBuffer()
- asyncio.create_task(delayBuf)
+ asyncio.create_task(delayBuf())
  return False
  try:
  self.dataChannel.send(msg)
  except Exception as e:
- log.error(f'DC#:${this.connectionId} Error when sending: {e}')
+ log.error(f'DC#:${self.connectionId} Error when sending: {e}')
  self._buffering = True
  self.close()
  return False
  return True
 
  def _tryBuffer(self) -> None:
  """Try to send the first message in the buffer."""
- if not this.open:
+ if not self.open:
  return
  if self._buffer.length == 0:
  return
 
- msg = this._buffer[0]
+ msg = self._buffer[0]
 
  if self._trySend(msg):
  self._buffer.shift()
  self._bufferSize = self._buffer.length
  self._tryBuffer()
 
- def _sendChunks(self, blob: Blob) -> None:
- blobs = util.chunk(blob)
- log.debug(f'DC#${this.connectionId} '
- f'Try to send ${blobs.length} chunks...')
- for blob in blobs:
- self.send(blob, True)
+ # def _sendChunks(self, blob: Blob) -> None:
+ #  blobs = util.chunk(blob)
+ #  log.debug(f'DC#${this.connectionId} '
+ #  f'Try to send ${blobs.length} chunks...')
+ #  for blob in blobs:
+ #  self.send(blob, True)
 
  def handleMessage(self, message: ServerMessage) -> None:
+ """Handle signaling server message."""
  payload = message.payload
  if message.type == ServerMessageType.Answer:
  self._negotiator.handleSDP(message.type, payload.sdp)

src/peerjs/socket.py

@@ -4,7 +4,7 @@
 from .enums import SocketEventType
 import json
 import websockets
-from websockets import WebSocket, ConnectionClosed
+from websockets.exceptions import ConnectionClosedError
 import asyncio
 
 log = logging.getLogger(__name__)
@@ -32,7 +32,7 @@ def __init__(
  self._disconnected: bool = True
  self._id: str = None
  self._messagesQueue: list = []
- self._websocket: WebSocket = None
+ self._websocket: websockets.client.WebSocketClientProtocol = None
  self._receiver: asyncio.Task = None
 
  async def _connect(self, wss_url=None):
@@ -61,8 +61,8 @@ async def _receive(self, websocket=None):
  log.warning("Invalid server message: {}, error {}",
  message, e)
  await self.emit('message', message)
- except ConnectionClosed:
- log.debug("Websocket connection closed.")
+ except ConnectionClosedError as err:
+ log.warning("Websocket connection closed with error. %s", err)
  except RuntimeError as e:
  log.warning("Websocket connection error: {}", e)
  finally: