You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

82 lines
2.0 KiB

  1. import WebSocketClient from 'websocket.js';
  2. const randomIntUpTo = max => Math.floor(Math.random() * Math.floor(max));
  3. export function connectStream(path, pollingRefresh = null, callbacks = () => ({ onDisconnect() {}, onReceive() {} })) {
  4. return (dispatch, getState) => {
  5. const streamingAPIBaseURL = getState().getIn(['meta', 'streaming_api_base_url']);
  6. const accessToken = getState().getIn(['meta', 'access_token']);
  7. const { onDisconnect, onReceive } = callbacks(dispatch, getState);
  8. let polling = null;
  9. const setupPolling = () => {
  10. pollingRefresh(dispatch, () => {
  11. polling = setTimeout(() => setupPolling(), 20000 + randomIntUpTo(20000));
  12. });
  13. };
  14. const clearPolling = () => {
  15. if (polling) {
  16. clearTimeout(polling);
  17. polling = null;
  18. }
  19. };
  20. const subscription = getStream(streamingAPIBaseURL, accessToken, path, {
  21. connected () {
  22. if (pollingRefresh) {
  23. clearPolling();
  24. }
  25. },
  26. disconnected () {
  27. if (pollingRefresh) {
  28. polling = setTimeout(() => setupPolling(), randomIntUpTo(40000));
  29. }
  30. onDisconnect();
  31. },
  32. received (data) {
  33. onReceive(data);
  34. },
  35. reconnected () {
  36. if (pollingRefresh) {
  37. clearPolling();
  38. pollingRefresh(dispatch);
  39. }
  40. },
  41. });
  42. const disconnect = () => {
  43. if (subscription) {
  44. subscription.close();
  45. }
  46. clearPolling();
  47. };
  48. return disconnect;
  49. };
  50. }
  51. export default function getStream(streamingAPIBaseURL, accessToken, stream, { connected, received, disconnected, reconnected }) {
  52. const params = [ `stream=${stream}` ];
  53. if (accessToken !== null) {
  54. params.push(`access_token=${accessToken}`);
  55. }
  56. const ws = new WebSocketClient(`${streamingAPIBaseURL}/api/v1/streaming/?${params.join('&')}`);
  57. ws.onopen = connected;
  58. ws.onmessage = e => received(JSON.parse(e.data));
  59. ws.onclose = disconnected;
  60. ws.onreconnect = reconnected;
  61. return ws;
  62. };