feat: datachannel handshake successful

- first successful datachannel message exchanged between ambianic-ui(PeerJS) and ambianic-edge(PeerJS-python)

Author: ivelin

src/peerjs/dataconnection.py

@@ -84,8 +84,8 @@ def _apply_options(
  # }
  # }
 
- self._dc: RTCDataChannel
- # self._encodingQueue = EncodingQueue()
+ self._dc: RTCDataChannel = None
+ self._encodingQueue = None # EncodingQueue()
  # @self._encodingQueue.on('done')
  # def on_eq_done(ab): # ab : ArrayBuffer
  # self._bufferedSend(ab)
@@ -101,10 +101,10 @@ def _apply_options(
  async def start(self):
  """Start data connection negotiation."""
  payload_options = self._payload or {'originator': True}
- log.info('Starting new connection with payload: %r '
- 'and payload_option: %r',
- self._payload,
- payload_options)
+ log.debug('\n Starting new connection with payload: \n %r '
+  '\n and payload_options: %r',
+  self._payload,
+  payload_options)
  await self._negotiator.startConnection(**payload_options)
 
  def initialize(self, dc: RTCDataChannel) -> None:
@@ -127,7 +127,9 @@ async def on_datachannel_open():
 
  @self.dataChannel.on('message')
  async def on_datachannel_message(e):
- log.debug(f'DC#${self.connectionId} dc onmessage: {e.data}')
+ log.info('DataChannel message from %s, \n: %r',
+ self.peerId,
+ e)
  await self._handleDataMessage(e)
 
  @self.dataChannel.on('close')

src/peerjs/negotiator.py

@@ -106,14 +106,21 @@ def _setupListeners(self, peerConnection: RTCPeerConnection = None):
  # });
  # };
 
+ @peerConnection.on('icegatheringstatechange')
+ def peerconn_icegatheringstatechange():
+ if peerConnection.iceGatheringState == 'complete':
+ log.warning('Local ICE candidates gathered.')
+
  @peerConnection.on('iceconnectionstatechange')
  def peerconn_oniceconnectionstatechange():
  state_change = BaseEventEmitter()
  @state_change.once("failed")
- def on_failed():
+ def on_failed(error=None):
  log.warning(
- f"iceConnectionState is failed, "
- "closing connections to {peerId}"
+ "\n IceConnectionState failed with error %s "
+ "\n Closing connections to peer id %s",
+ error,
+ peerId
  )
  self.connection.emit(
  ConnectionEventType.Error,
@@ -148,7 +155,7 @@ def on_disconnected():
 
  @state_change.once("completed")
  def on_completed():
- log.info(
+ log.warning(
  'iceConnectionState completed for peer id %s',
  peerId
  )
@@ -160,6 +167,10 @@ def on_completed():
  # peerconn_onicecandidate)
  pass
 
+ log.warning(
+ 'iceConnectionState event: %s',
+ peerConnection.iceConnectionState
+ )
  # forward connection stage change event to local handlers
  state_change.emit(peerConnection.iceConnectionState)
  # notify higher level connection listeners
@@ -171,9 +182,8 @@ def on_completed():
  # Fired between offer and answer, so options should already be saved
  # in the options hash.
  @peerConnection.on("datachannel")
- def peerconn_ondatachanel(evt):
- log.info("Received data channel.")
- dataChannel = evt.channel
+ def peerconn_ondatachanel(dataChannel):
+ log.warning("Received data channel %r", dataChannel)
  connection = provider.getConnection(peerId, connectionId)
  connection.initialize(dataChannel)
 
@@ -207,7 +217,10 @@ async def cleanup(self) -> None:
  dataChannelNotClosed = dataChannel.readyState and \
  dataChannel.readyState != "closed"
  if peerConnectionNotClosed or dataChannelNotClosed:
- await peerConnection.close()
+ try:
+ await peerConnection.close()
+ except Exception as err:
+ log.exception('Error while closing connection', err)
 
  async def _makeOffer(self):
  peerConnection = self.connection.peerConnection
@@ -262,21 +275,23 @@ async def _makeAnswer(self) -> None:
  provider = self.connection.provider
  try:
  answer = await peerConnection.createAnswer()
- log.info('\n Created answer: \n %r', answer)
- log.info('\n Connection options: %r',
- self.connection.options)
  sdpTransformFunction = \
  self.connection.options.get('sdpTransform', None)
  if sdpTransformFunction and \
  callable(sdpTransformFunction):
  answer.sdp = sdpTransformFunction(answer.sdp) \
  or answer.sdp
- json_answer = object_to_string(answer)
+ log.info('\n Created answer header: \n %r', answer)
+ log.info('\n Connection options: %r',
+ self.connection.options)
  try:
+ log.info('Gathering ICE candidates to complete answer...')
  await peerConnection.setLocalDescription(answer)
- log.info(f"Set localDescription: {answer}"
- f"for:{self.connection.peer}")
-
+ answer = peerConnection.localDescription
+ json_answer = object_to_string(answer)
+ log.warning('\n Sending SDP ANSWER to peer id %s: \n %r ',
+ self.connection.peer,
+ json_answer)
  await provider.socket.send({
  'type': ServerMessageType.Answer,
  'payload': {
@@ -299,12 +314,12 @@ async def handleSDP(self, type: str = None, sdp: str = None) -> None:
  rsd = RTCSessionDescription(type=type, sdp=sdp)
  peerConnection = self.connection.peerConnection
  provider = self.connection.provider
- log.info("Setting remote session description %r", rsd)
+ log.debug("\n Setting remote session description \n %r", rsd)
  try:
  await peerConnection.setRemoteDescription(rsd)
- log.info('Set remoteDescription type: %s'
- 'for peer: %r',
- type, self.connection.peer)
+ log.debug('\n Set remoteDescription type: %s'
+  '\n for peer: %r',
+  type, self.connection.peer)
  if type == "offer":
  await self._makeAnswer()
  except Exception as err:

src/peerjs/peer.py

@@ -198,9 +198,9 @@ async def _handleMessage(self, message: ServerMessage) -> None:
  peerId = message.src
  payload = message.payload
 
- log.info('\n Handling server message \n type %s, '
- '\n source peer/client id %s, \n payload %s, \n message %r',
- type, peerId, payload, message)
+ log.debug('\n Handling server message \n type %s, '
+  '\n source peer/client id %s, \n payload %s, \n message %r',
+  type, peerId, payload, message)
 
  server_messenger = AsyncIOEventEmitter()
 
@@ -273,8 +273,8 @@ async def _handle_offer(self, peerId=None, payload=None):
  """Handle remote peer offer for a direct connection."""
  # we should consider switching this to CALL/CONNECT,
  # but this is the least breaking option.
- log.info("Remote peer id %s offering connection. Payload %s",
- peerId, payload)
+ log.debug("\n Remote peer id %s offering connection. \n Payload %s",
+  peerId, payload)
  connectionId = payload['connectionId']
  connection = self.getConnection(peerId, connectionId)