add ability to stream statuses

package-lock.json

@@ -264,6 +264,14 @@
         "js-tokens": "3.0.2"
       }
     },
+    "backoff": {
+      "version": "2.5.0",
+      "resolved": "https://registry.npmjs.org/backoff/-/backoff-2.5.0.tgz",
+      "integrity": "sha1-9hbtqdPktmuMp/ynn2lXIsX44m8=",
+      "requires": {
+        "precond": "0.2.3"
+      }
+    },
     "balanced-match": {
       "version": "0.4.2",
       "resolved": "https://registry.npmjs.org/balanced-match/-/balanced-match-0.4.2.tgz",
@@ -5473,6 +5481,11 @@
         "uniqs": "2.0.0"
       }
     },
+    "precond": {
+      "version": "0.2.3",
+      "resolved": "https://registry.npmjs.org/precond/-/precond-0.2.3.tgz",
+      "integrity": "sha1-qpWRvKokkj8eD0hJ0kD0fvwQdaw="
+    },
     "prepend-http": {
       "version": "1.0.4",
       "resolved": "https://registry.npmjs.org/prepend-http/-/prepend-http-1.0.4.tgz",
@@ -7779,6 +7792,14 @@
         }
       }
     },
+    "websocket.js": {
+      "version": "0.1.12",
+      "resolved": "https://registry.npmjs.org/websocket.js/-/websocket.js-0.1.12.tgz",
+      "integrity": "sha1-RsmAeHxX68jtz0SgJj5dY5NnuFs=",
+      "requires": {
+        "backoff": "2.5.0"
+      }
+    },
     "whet.extend": {
       "version": "0.9.9",
       "resolved": "https://registry.npmjs.org/whet.extend/-/whet.extend-0.9.9.tgz",

package.json

@@ -52,6 +52,7 @@
     "url-search-params": "^0.10.0",
     "webpack": "^3.10.0",
     "webpack-bundle-analyzer": "^2.9.2",
+    "websocket.js": "^0.1.12",
     "workerize-loader": "^1.0.1",
     "yargs": "^10.1.1"
   },

routes/_components/Timeline.html

@@ -29,6 +29,7 @@
   import { timelines } from '../_static/timelines'
   import { toast } from '../_utils/toast'
   import { database } from '../_utils/database/database'
+  import { StatusStream } from '../_utils/mastodon/StatusStream'
 
   const cachedTimelines = {}
 
@@ -42,6 +43,7 @@
     async oncreate() {
       let timeline = this.get('timeline')
       let instanceName = this.store.get('currentInstance')
+      let accessToken = this.store.get('accessToken')
       let cachedStatusIds = cachedTimelines[timeline]
       if (cachedStatusIds) {
         this.set({statusIds: cachedStatusIds})
@@ -50,8 +52,16 @@
       }
       /* no await */ getInstanceInfo(instanceName).then(instanceInfo => database.setInstanceInfo(instanceName, instanceInfo))
       let instanceInfo = await database.getInstanceInfo(instanceName)
+      this._statusStream = new StatusStream(instanceInfo.urls.streaming_api, accessToken, timeline, {
+        onMessage(message) {
+          console.log('message', message)
+        }
+      })
     },
     ondestroy() {
+      if (this._statusStream) {
+        this._statusStream.close()
+      }
       cachedTimelines[this.get('timeline')] = this.get('statusIds')
     },
     data: () => ({

routes/_utils/asyncModules.js

@@ -1,6 +1,6 @@
 import { loadCSS } from 'fg-loadcss';
 
-const importURLSearchParams = () => import(
+export const importURLSearchParams = () => import(
   /* webpackChunkName: 'url-search-params' */ 'url-search-params'
   ).then(Params => {
   window.URLSearchParams = Params
@@ -11,23 +11,23 @@ const importURLSearchParams = () => import(
   })
 })
 
-const importTimeline = () => import(
+export const importTimeline = () => import(
   /* webpackChunkName: 'Timeline' */ '../_components/Timeline.html'
   ).then(mod => mod.default)
 
-const importIntersectionObserver = () => import(
+export const importIntersectionObserver = () => import(
   /* webpackChunkName: 'intersection-observer' */ 'intersection-observer'
   )
 
-const importRequestIdleCallback = () => import(
+export const importRequestIdleCallback = () => import(
   /* webpackChunkName: 'requestidlecallback' */ 'requestidlecallback'
   )
 
-const importIndexedDBGetAllShim = () => import(
+export const importIndexedDBGetAllShim = () => import(
   /* webpackChunkName: 'indexeddb-getall-shim' */ 'indexeddb-getall-shim'
   )
 
-const importDialogPolyfill = (() => {
+export const importDialogPolyfill = (() => {
   let cached
   return () => {
     if (cached) {
@@ -39,13 +39,4 @@ const importDialogPolyfill = (() => {
       return cached
     })
   }
-})()
-
-export {
-  importURLSearchParams,
-  importTimeline,
-  importIntersectionObserver,
-  importRequestIdleCallback,
-  importIndexedDBGetAllShim,
-  importDialogPolyfill
-}
+})()

routes/_utils/mastodon/StatusStream.js

@@ -0,0 +1,58 @@
+import { paramsString } from '../ajax'
+import noop from 'lodash/noop'
+import WebSocketClient from 'websocket.js'
+
+function getStreamName(timeline) {
+  switch (timeline) {
+    case 'local':
+      return 'public:local'
+    case 'federated':
+      return 'public'
+    case 'home':
+      return 'user'
+    case 'notifications':
+      return 'user:notification'
+  }
+  if (timeline.startsWith('tag/')) {
+    return 'hashtag'
+  }
+}
+
+function getUrl(streamingApi, accessToken, timeline) {
+  let url = `${streamingApi}/api/v1/streaming`
+  let streamName = getStreamName(timeline)
+
+  let params = {
+    stream: streamName
+  }
+
+  if (timeline.startsWith('tag/')) {
+    params.tag = timeline.split('/').slice(-1)[0]
+  }
+
+  if (accessToken) {
+    params.access_token = accessToken
+  }
+
+  return url + '?' + paramsString(params)
+}
+
+export class StatusStream {
+  constructor(streamingApi, accessToken, timeline, opts) {
+    let url = getUrl(streamingApi, accessToken, timeline)
+
+    const ws = new WebSocketClient(url, null, { backoff: 'exponential' })
+    const onMessage = opts.onMessage || noop
+
+    ws.onopen = opts.onOpen || noop
+    ws.onmessage = e => onMessage(JSON.parse(e.data))
+    ws.onclose = opts.onClose || noop
+    ws.onreconnect = opts.onReconnect || noop
+
+    this._ws = ws
+  }
+
+  close() {
+    this._ws.close()
+  }
+}