Make the streaming API also handle websockets (because trying to get …

…the browser EventSource interface to

work flawlessly was a nightmare). WARNING: This commit makes the web UI connect to the streaming API instead
of ActionCable like before. This means that if you are upgrading, you should set that up beforehand.

.env.production.sample

@@ -43,5 +43,5 @@ [email protected]
 # Optional alias for S3 if you want to use Cloudfront or Cloudflare in front
 # S3_CLOUDFRONT_HOST=
 
-# Optional Firebase Cloud Messaging API key
-FCM_API_KEY=
+# Streaming API integration
+# STREAMING_API_BASE_URL=

app/assets/javascripts/application.js

@@ -13,4 +13,3 @@
 //= require jquery
 //= require jquery_ujs
 //= require components
-//= require cable

app/assets/javascripts/components/containers/mastodon.jsx

@@ -43,6 +43,7 @@ import hu from 'react-intl/locale-data/hu';
 import uk from 'react-intl/locale-data/uk';
 import getMessagesForLocale from '../locales';
 import { hydrateStore } from '../actions/store';
+import createStream from '../stream';
 
 const store = configureStore();
 
@@ -60,28 +61,27 @@ const Mastodon = React.createClass({
  locale: React.PropTypes.string.isRequired
  },
 
- componentWillMount() {
- const { locale } = this.props;
-
- if (typeof App !== 'undefined') {
- this.subscription = App.cable.subscriptions.create('TimelineChannel', {
-
- received (data) {
- switch(data.event) {
- case 'update':
- store.dispatch(updateTimeline('home', JSON.parse(data.payload)));
- break;
- case 'delete':
- store.dispatch(deleteFromTimelines(data.payload));
- break;
- case 'notification':
- store.dispatch(updateNotifications(JSON.parse(data.payload), getMessagesForLocale(locale), locale));
- break;
- }
+ componentDidMount() {
+ const { locale } = this.props;
+ const accessToken = store.getState().getIn(['meta', 'access_token']);
+
+ this.subscription = createStream(accessToken, 'user', {
+
+ received (data) {
+ switch(data.event) {
+ case 'update':
+ store.dispatch(updateTimeline('home', JSON.parse(data.payload)));
+ break;
+ case 'delete':
+ store.dispatch(deleteFromTimelines(data.payload));
+ break;
+ case 'notification':
+ store.dispatch(updateNotifications(JSON.parse(data.payload), getMessagesForLocale(locale), locale));
+ break;
  }
+ }
 
- });
- }
+ });
 
  // Desktop notifications
  if (typeof window.Notification !== 'undefined' && Notification.permission === 'default') {
@@ -91,7 +91,8 @@ const Mastodon = React.createClass({
 
  componentWillUnmount () {
  if (typeof this.subscription !== 'undefined') {
- this.subscription.unsubscribe();
+ this.subscription.close();
+ this.subscription = null;
  }
  },
 

app/assets/javascripts/components/features/hashtag_timeline/index.jsx

@@ -8,45 +8,49 @@ import {
  deleteFromTimelines
 } from '../../actions/timelines';
 import ColumnBackButtonSlim from '../../components/column_back_button_slim';
+import createStream from '../../stream';
+
+const mapStateToProps = state => ({
+ accessToken: state.getIn(['meta', 'access_token'])
+});
 
 const HashtagTimeline = React.createClass({
 
  propTypes: {
  params: React.PropTypes.object.isRequired,
- dispatch: React.PropTypes.func.isRequired
+ dispatch: React.PropTypes.func.isRequired,
+ accessToken: React.PropTypes.string.isRequired
  },
 
  mixins: [PureRenderMixin],
 
  _subscribe (dispatch, id) {
- if (typeof App !== 'undefined') {
- this.subscription = App.cable.subscriptions.create({
- channel: 'HashtagChannel',
- tag: id
- }, {
-
- received (data) {
- switch(data.event) {
- case 'update':
- dispatch(updateTimeline('tag', JSON.parse(data.payload)));
- break;
- case 'delete':
- dispatch(deleteFromTimelines(data.payload));
- break;
- }
+ const { accessToken } = this.props;
+
+ this.subscription = createStream(accessToken, `hashtag&tag=${id}`, {
+
+ received (data) {
+ switch(data.event) {
+ case 'update':
+ dispatch(updateTimeline('tag', JSON.parse(data.payload)));
+ break;
+ case 'delete':
+ dispatch(deleteFromTimelines(data.payload));
+ break;
  }
+ }
 
- });
- }
+ });
  },
 
  _unsubscribe () {
  if (typeof this.subscription !== 'undefined') {
- this.subscription.unsubscribe();
+ this.subscription.close();
+ this.subscription = null;
  }
  },
 
- componentWillMount () {
+ componentDidMount () {
  const { dispatch } = this.props;
  const { id } = this.props.params;
 
@@ -79,4 +83,4 @@ const HashtagTimeline = React.createClass({
 
 });
 
-export default connect()(HashtagTimeline);
+export default connect(mapStateToProps)(HashtagTimeline);

app/assets/javascripts/components/features/public_timeline/index.jsx

@@ -9,46 +9,51 @@ import {
 } from '../../actions/timelines';
 import { defineMessages, injectIntl } from 'react-intl';
 import ColumnBackButtonSlim from '../../components/column_back_button_slim';
+import createStream from '../../stream';
 
 const messages = defineMessages({
  title: { id: 'column.public', defaultMessage: 'Public' }
 });
 
+const mapStateToProps = state => ({
+ accessToken: state.getIn(['meta', 'access_token'])
+});
+
 const PublicTimeline = React.createClass({
 
  propTypes: {
  dispatch: React.PropTypes.func.isRequired,
- intl: React.PropTypes.object.isRequired
+ intl: React.PropTypes.object.isRequired,
+ accessToken: React.PropTypes.string.isRequired
  },
 
  mixins: [PureRenderMixin],
 
- componentWillMount () {
- const { dispatch } = this.props;
+ componentDidMount () {
+ const { dispatch, accessToken } = this.props;
 
  dispatch(refreshTimeline('public'));
 
- if (typeof App !== 'undefined') {
- this.subscription = App.cable.subscriptions.create('PublicChannel', {
+ this.subscription = createStream(accessToken, 'public', {
 
- received (data) {
- switch(data.event) {
- case 'update':
- dispatch(updateTimeline('public', JSON.parse(data.payload)));
- break;
- case 'delete':
- dispatch(deleteFromTimelines(data.payload));
- break;
- }
+ received (data) {
+ switch(data.event) {
+ case 'update':
+ dispatch(updateTimeline('public', JSON.parse(data.payload)));
+ break;
+ case 'delete':
+ dispatch(deleteFromTimelines(data.payload));
+ break;
  }
+ }
 
- });
- }
+ });
  },
 
  componentWillUnmount () {
  if (typeof this.subscription !== 'undefined') {
- this.subscription.unsubscribe();
+ this.subscription.close();
+ this.subscription = null;
  }
  },
 
@@ -65,4 +70,4 @@ const PublicTimeline = React.createClass({
 
 });
 
-export default connect()(injectIntl(PublicTimeline));
+export default connect(mapStateToProps)(injectIntl(PublicTimeline));

app/assets/javascripts/components/stream.jsx

@@ -0,0 +1,21 @@
+import WebSocketClient from 'websocket.js';
+
+const createWebSocketURL = (url) => {
+ const a = document.createElement('a');
+
+ a.href = url;
+ a.href = a.href;
+ a.protocol = a.protocol.replace('http', 'ws');
+
+ return a.href;
+};
+
+export default function getStream(accessToken, stream, { connected, received, disconnected }) {
+ const ws = new WebSocketClient(`${createWebSocketURL(STREAMING_API_BASE_URL)}/api/v1/streaming/?access_token=${accessToken}&stream=${stream}`);
+
+ ws.onopen = connected;
+ ws.onmessage = e => received(JSON.parse(e.data));
+ ws.onclose = disconnected;
+
+ return ws;
+};

app/views/home/index.html.haml

@@ -1,5 +1,6 @@
 - content_for :header_tags do
  :javascript
+ window.STREAMING_API_BASE_URL = '#{Rails.configuration.x.streaming_api_base_url}';
  window.INITIAL_STATE = #{json_escape(render(file: 'home/initial_state', formats: :json))}
 
  = javascript_include_tag 'application'

config/initializers/ostatus.rb

@@ -10,8 +10,10 @@
  config.x.use_s3 = ENV['S3_ENABLED'] == 'true'
 
  config.action_mailer.default_url_options = { host: host, protocol: https ? 'https://' : 'http://', trailing_slash: false }
+ config.x.streaming_api_base_url = 'http://localhost:4000'
 
  if Rails.env.production?
  config.action_cable.allowed_request_origins = ["http#{https ? 's' : ''}://#{host}"]
+ config.x.streaming_api_base_url = ENV.fetch('STREAMING_API_BASE_URL') { "http#{https ? 's' : ''}://#{host}" }
  end
 end

docker-compose.yml

@@ -19,6 +19,16 @@ services:
  volumes:
  - ./public/assets:/mastodon/public/assets
  - ./public/system:/mastodon/public/system
+ streaming:
+ restart: always
+ build: .
+ env_file: .env.production
+ command: npm run start
+ ports:
+ - "4000:4000"
+ depends_on:
+ - db
+ - redis
  sidekiq:
  restart: always
  build: .

docs/Running-Mastodon/Production-guide.md

@@ -49,6 +49,22 @@ server {
  tcp_nodelay on;
  }
 
+ location /api/v1/streaming {
+ proxy_set_header Host $host;
+ proxy_set_header X-Real-IP $remote_addr;
+ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+ proxy_set_header X-Forwarded-Proto https;
+
+ proxy_pass http://localhost:4000;
+ proxy_buffering off;
+ proxy_redirect off;
+ proxy_http_version 1.1;
+ proxy_set_header Upgrade $http_upgrade;
+ proxy_set_header Connection $connection_upgrade;
+
+ tcp_nodelay on;
+ }
+
  error_page 500 501 502 503 504 /500.html;
 }
 ```
@@ -162,6 +178,27 @@ Restart=always
 WantedBy=multi-user.target
 ```
 
+Example systemd configuration file for the streaming API, to be placed in `/etc/systemd/system/mastodon-streaming.service`:
+
+```systemd
+[Unit]
+Description=mastodon-streaming
+After=network.target
+
+[Service]
+Type=simple
+User=mastodon
+WorkingDirectory=/home/mastodon/live
+Environment="NODE_ENV=production"
+Environment="PORT=4000"
+ExecStart=/usr/bin/npm run start
+TimeoutSec=15
+Restart=always
+
+[Install]
+WantedBy=multi-user.target
+```
+
 This allows you to `sudo systemctl enable mastodon-*.service` and `sudo systemctl start mastodon-*.service` to get things going.
 
 ## Cronjobs

package.json

@@ -18,6 +18,7 @@
  "babelify": "^7.3.0",
  "browserify": "^13.1.0",
  "browserify-incremental": "^3.1.1",
+ "bufferutil": "^2.0.0",
  "chai": "^3.5.0",
  "chai-enzyme": "^0.5.2",
  "css-loader": "^0.26.1",
@@ -64,6 +65,9 @@
  "sass-loader": "^4.0.2",
  "sinon": "^1.17.6",
  "style-loader": "^0.13.1",
- "webpack": "^1.14.0"
+ "utf-8-validate": "^3.0.0",
+ "webpack": "^1.14.0",
+ "websocket.js": "^0.1.7",
+ "ws": "^2.0.2"
  }
 }