dataconnection.ts 8.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333
  1. import { Reliable } from "reliable";
  2. import { util } from "./util";
  3. import logger, { LogLevel } from "./logger";
  4. import { Negotiator } from "./negotiator";
  5. import {
  6. ConnectionType,
  7. ConnectionEventType,
  8. SerializationType,
  9. ServerMessageType
  10. } from "./enums";
  11. import { Peer } from "./peer";
  12. import { BaseConnection } from "./baseconnection";
  13. import { ServerMessage } from "./servermessage";
  14. /**
  15. * Wraps a DataChannel between two Peers.
  16. */
  17. export class DataConnection extends BaseConnection {
  18. private static readonly ID_PREFIX = "dc_";
  19. private _negotiator: Negotiator;
  20. readonly label: string;
  21. readonly serialization: SerializationType;
  22. readonly reliable: boolean;
  23. get type() {
  24. return ConnectionType.Data;
  25. }
  26. private _buffer: any[] = [];
  27. private _bufferSize = 0;
  28. private _buffering = false;
  29. private _chunkedData = {};
  30. private _peerBrowser: any;
  31. private _dc: RTCDataChannel;
  32. private _reliable: Reliable;
  33. get dataChannel(): RTCDataChannel {
  34. return this._dc;
  35. }
  36. get bufferSize(): number { return this._bufferSize; }
  37. constructor(peerId: string, provider: Peer, options: any) {
  38. super(peerId, provider, options);
  39. this.connectionId =
  40. this.options.connectionId || DataConnection.ID_PREFIX + util.randomToken();
  41. this.label = this.options.label || this.connectionId;
  42. this.serialization = this.options.serialization || SerializationType.Binary;
  43. this.reliable = this.options.reliable;
  44. if (this.options._payload) {
  45. this._peerBrowser = this.options._payload.browser;
  46. }
  47. this._negotiator = new Negotiator(this);
  48. this._negotiator.startConnection(
  49. this.options._payload || {
  50. originator: true
  51. }
  52. );
  53. }
  54. /** Called by the Negotiator when the DataChannel is ready. */
  55. initialize(dc: RTCDataChannel): void {
  56. this._dc = dc;
  57. this._configureDataChannel();
  58. }
  59. private _configureDataChannel(): void {
  60. if (util.supports.sctp) {
  61. this.dataChannel.binaryType = "arraybuffer";
  62. }
  63. this.dataChannel.onopen = () => {
  64. logger.log("Data channel connection success");
  65. this._open = true;
  66. this.emit(ConnectionEventType.Open);
  67. };
  68. // Use the Reliable shim for non Firefox browsers
  69. if (!util.supports.sctp && this.reliable) {
  70. const isLoggingEnable = logger.logLevel > LogLevel.Disabled;
  71. this._reliable = new Reliable(this.dataChannel, isLoggingEnable);
  72. }
  73. if (this._reliable) {
  74. this._reliable.onmessage = (msg) => {
  75. this.emit(ConnectionEventType.Data, msg);
  76. };
  77. } else {
  78. this.dataChannel.onmessage = (e) => {
  79. this._handleDataMessage(e);
  80. };
  81. }
  82. this.dataChannel.onclose = () => {
  83. logger.log("DataChannel closed for:", this.peer);
  84. this.close();
  85. };
  86. }
  87. // Handles a DataChannel message.
  88. private _handleDataMessage({ data }): void {
  89. const datatype = data.constructor;
  90. const isBinarySerialization = this.serialization === SerializationType.Binary ||
  91. this.serialization === SerializationType.BinaryUTF8;
  92. let deserializedData = data;
  93. if (isBinarySerialization) {
  94. if (datatype === Blob) {
  95. // Datatype should never be blob
  96. util.blobToArrayBuffer(data, (ab) => {
  97. const unpackedData = util.unpack(ab);
  98. this.emit(ConnectionEventType.Data, unpackedData);
  99. });
  100. return;
  101. } else if (datatype === ArrayBuffer) {
  102. deserializedData = util.unpack(data);
  103. } else if (datatype === String) {
  104. // String fallback for binary data for browsers that don't support binary yet
  105. const ab = util.binaryStringToArrayBuffer(data);
  106. deserializedData = util.unpack(ab);
  107. }
  108. } else if (this.serialization === SerializationType.JSON) {
  109. deserializedData = JSON.parse(data);
  110. }
  111. // Check if we've chunked--if so, piece things back together.
  112. // We're guaranteed that this isn't 0.
  113. if (deserializedData.__peerData) {
  114. this._handleChunk(deserializedData);
  115. return;
  116. }
  117. super.emit(ConnectionEventType.Data, deserializedData);
  118. }
  119. private _handleChunk(data: any): void {
  120. const id = data.__peerData;
  121. const chunkInfo = this._chunkedData[id] || {
  122. data: [],
  123. count: 0,
  124. total: data.total
  125. };
  126. chunkInfo.data[data.n] = data.data;
  127. chunkInfo.count++;
  128. this._chunkedData[id] = chunkInfo;
  129. if (chunkInfo.total === chunkInfo.count) {
  130. // Clean up before making the recursive call to `_handleDataMessage`.
  131. delete this._chunkedData[id];
  132. // We've received all the chunks--time to construct the complete data.
  133. const data = new Blob(chunkInfo.data);
  134. this._handleDataMessage({ data });
  135. }
  136. }
  137. /**
  138. * Exposed functionality for users.
  139. */
  140. /** Allows user to close connection. */
  141. close(): void {
  142. this._buffer = [];
  143. this._bufferSize = 0;
  144. this._chunkedData = {};
  145. if (this._negotiator) {
  146. this._negotiator.cleanup();
  147. this._negotiator = null;
  148. }
  149. if (this.provider) {
  150. this.provider._removeConnection(this);
  151. this.provider = null;
  152. }
  153. if (!this.open) {
  154. return;
  155. }
  156. this._open = false;
  157. super.emit(ConnectionEventType.Close);
  158. }
  159. /** Allows user to send data. */
  160. send(data: any, chunked: boolean): void {
  161. if (!this.open) {
  162. super.emit(
  163. ConnectionEventType.Error,
  164. new Error(
  165. "Connection is not open. You should listen for the `open` event before sending messages."
  166. )
  167. );
  168. return;
  169. }
  170. if (this._reliable) {
  171. // Note: reliable shim sending will make it so that you cannot customize
  172. // serialization.
  173. this._reliable.send(data);
  174. return;
  175. }
  176. if (this.serialization === SerializationType.JSON) {
  177. this._bufferedSend(JSON.stringify(data));
  178. } else if (
  179. this.serialization === SerializationType.Binary ||
  180. this.serialization === SerializationType.BinaryUTF8
  181. ) {
  182. const blob = util.pack(data);
  183. // For Chrome-Firefox interoperability, we need to make Firefox "chunk"
  184. // the data it sends out.
  185. const needsChunking =
  186. util.chunkedBrowsers[this._peerBrowser] ||
  187. util.chunkedBrowsers[util.browser];
  188. if (needsChunking && !chunked && blob.size > util.chunkedMTU) {
  189. this._sendChunks(blob);
  190. return;
  191. }
  192. // DataChannel currently only supports strings.
  193. if (!util.supports.sctp) {
  194. util.blobToBinaryString(blob, (str) => {
  195. this._bufferedSend(str);
  196. });
  197. } else if (!util.supports.binaryBlob) {
  198. // We only do this if we really need to (e.g. blobs are not supported),
  199. // because this conversion is costly.
  200. util.blobToArrayBuffer(blob, (ab) => {
  201. this._bufferedSend(ab);
  202. });
  203. } else {
  204. this._bufferedSend(blob);
  205. }
  206. } else {
  207. this._bufferedSend(data);
  208. }
  209. }
  210. private _bufferedSend(msg: any): void {
  211. if (this._buffering || !this._trySend(msg)) {
  212. this._buffer.push(msg);
  213. this._bufferSize = this._buffer.length;
  214. }
  215. }
  216. // Returns true if the send succeeds.
  217. private _trySend(msg: any): boolean {
  218. if (!this.open) {
  219. return false;
  220. }
  221. try {
  222. this.dataChannel.send(msg);
  223. } catch (e) {
  224. this._buffering = true;
  225. setTimeout(() => {
  226. // Try again.
  227. this._buffering = false;
  228. this._tryBuffer();
  229. }, 100);
  230. return false;
  231. }
  232. return true;
  233. }
  234. // Try to send the first message in the buffer.
  235. private _tryBuffer(): void {
  236. if (!this.open) {
  237. return;
  238. }
  239. if (this._buffer.length === 0) {
  240. return;
  241. }
  242. const msg = this._buffer[0];
  243. if (this._trySend(msg)) {
  244. this._buffer.shift();
  245. this._bufferSize = this._buffer.length;
  246. this._tryBuffer();
  247. }
  248. }
  249. private _sendChunks(blob): void {
  250. const blobs = util.chunk(blob);
  251. for (let blob of blobs) {
  252. this.send(blob, true);
  253. }
  254. }
  255. handleMessage(message: ServerMessage): void {
  256. const payload = message.payload;
  257. switch (message.type) {
  258. case ServerMessageType.Answer:
  259. this._peerBrowser = payload.browser;
  260. // Forward to negotiator
  261. this._negotiator.handleSDP(message.type, payload.sdp);
  262. break;
  263. case ServerMessageType.Candidate:
  264. this._negotiator.handleCandidate(payload.candidate);
  265. break;
  266. default:
  267. logger.warn(
  268. "Unrecognized message type:",
  269. message.type,
  270. "from peer:",
  271. this.peer
  272. );
  273. break;
  274. }
  275. }
  276. }