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.
# 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=
//= require jquery
//= require jquery_ujs
//= require components
-//= require cable
+++ /dev/null
-// Action Cable provides the framework to deal with WebSockets in Rails.
-// You can generate new channels where WebSocket features live using the rails generate channel command.
-//
-//= require action_cable
-//= require_self
-
-(function() {
- this.App || (this.App = {});
-
- App.cable = ActionCable.createConsumer();
-
-}).call(this);
import uk from 'react-intl/locale-data/uk';
import getMessagesForLocale from '../locales';
import { hydrateStore } from '../actions/store';
+import createStream from '../stream';
const store = configureStore();
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') {
componentWillUnmount () {
if (typeof this.subscription !== 'undefined') {
- this.subscription.unsubscribe();
+ this.subscription.close();
+ this.subscription = null;
}
},
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;
});
-export default connect()(HashtagTimeline);
+export default connect(mapStateToProps)(HashtagTimeline);
} 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;
}
},
});
-export default connect()(injectIntl(PublicTimeline));
+export default connect(mapStateToProps)(injectIntl(PublicTimeline));
--- /dev/null
+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;
+};
- 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.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
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: .
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;
}
```
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
"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",
"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"
}
}
import dotenv from 'dotenv'
import express from 'express'
+import http from 'http'
import redis from 'redis'
import pg from 'pg'
import log from 'npmlog'
+import url from 'url'
+import WebSocket from 'ws'
const env = process.env.NODE_ENV || 'development'
}
}
-const app = express()
+const app = express()
const pgPool = new pg.Pool(pgConfigs[env])
+const server = http.createServer(app)
+const wss = new WebSocket.Server({ server })
const allowCrossDomain = (req, res, next) => {
res.header('Access-Control-Allow-Origin', '*')
next()
}
-const authenticationMiddleware = (req, res, next) => {
- if (req.method === 'OPTIONS') {
- return next()
- }
-
- const authorization = req.get('Authorization')
-
- if (!authorization) {
- const err = new Error('Missing access token')
- err.statusCode = 401
-
- return next(err)
- }
-
- const token = authorization.replace(/^Bearer /, '')
-
+const accountFromToken = (token, req, next) => {
pgPool.connect((err, client, done) => {
if (err) {
return next(err)
})
}
+const authenticationMiddleware = (req, res, next) => {
+ if (req.method === 'OPTIONS') {
+ return next()
+ }
+
+ const authorization = req.get('Authorization')
+
+ if (!authorization) {
+ const err = new Error('Missing access token')
+ err.statusCode = 401
+
+ return next(err)
+ }
+
+ const token = authorization.replace(/^Bearer /, '')
+
+ accountFromToken(token, req, next)
+}
+
const errorMiddleware = (err, req, res, next) => {
log.error(err)
res.writeHead(err.statusCode || 500, { 'Content-Type': 'application/json' })
- res.end(JSON.stringify({ error: err.statusCode ? `${err}` : 'An unexpected error occured' }))
+ res.end(JSON.stringify({ error: err.statusCode ? `${err}` : 'An unexpected error occurred' }))
}
const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
-const streamFrom = (id, req, res, needsFiltering = false) => {
+const streamFrom = (redisClient, id, req, output, needsFiltering = false) => {
log.verbose(`Starting stream from ${id} for ${req.accountId}`)
- res.setHeader('Content-Type', 'text/event-stream')
- res.setHeader('Transfer-Encoding', 'chunked')
-
- const redisClient = redis.createClient({
- host: process.env.REDIS_HOST || '127.0.0.1',
- port: process.env.REDIS_PORT || 6379,
- password: process.env.REDIS_PASSWORD
- })
-
redisClient.on('message', (channel, message) => {
const { event, payload } = JSON.parse(message)
return
}
- res.write(`event: ${event}\n`)
- res.write(`data: ${payload}\n\n`)
+ log.silly(`Transmitting for ${req.accountId}: ${event} ${payload}`)
+ output(event, payload)
})
})
} else {
- res.write(`event: ${event}\n`)
- res.write(`data: ${payload}\n\n`)
+ log.silly(`Transmitting for ${req.accountId}: ${event} ${payload}`)
+ output(event, payload)
}
})
+ redisClient.subscribe(id)
+}
+
+// Setup stream output to HTTP
+const streamToHttp = (req, res, redisClient) => {
+ res.setHeader('Content-Type', 'text/event-stream')
+ res.setHeader('Transfer-Encoding', 'chunked')
+
const heartbeat = setInterval(() => res.write(':thump\n'), 15000)
req.on('close', () => {
- log.verbose(`Ending stream from ${id} for ${req.accountId}`)
+ log.verbose(`Ending stream for ${req.accountId}`)
clearInterval(heartbeat)
redisClient.quit()
})
- redisClient.subscribe(id)
+ return (event, payload) => {
+ res.write(`event: ${event}\n`)
+ res.write(`data: ${payload}\n\n`)
+ }
+}
+
+// Setup stream output to WebSockets
+const streamToWs = (req, ws, redisClient) => {
+ ws.on('close', () => {
+ log.verbose(`Ending stream for ${req.accountId}`)
+ redisClient.quit()
+ })
+
+ return (event, payload) => {
+ ws.send(JSON.stringify({ event, payload }))
+ }
}
+// Get new redis connection
+const getRedisClient = () => redis.createClient({
+ host: process.env.REDIS_HOST || '127.0.0.1',
+ port: process.env.REDIS_PORT || 6379,
+ password: process.env.REDIS_PASSWORD
+})
+
app.use(allowCrossDomain)
app.use(authenticationMiddleware)
app.use(errorMiddleware)
-app.get('/api/v1/streaming/user', (req, res) => streamFrom(`timeline:${req.accountId}`, req, res))
-app.get('/api/v1/streaming/public', (req, res) => streamFrom('timeline:public', req, res, true))
-app.get('/api/v1/streaming/hashtag', (req, res) => streamFrom(`timeline:hashtag:${req.params.tag}`, req, res, true))
+app.get('/api/v1/streaming/user', (req, res) => {
+ const redisClient = getRedisClient()
+ streamFrom(redisClient, `timeline:${req.accountId}`, req, streamToHttp(req, res, redisClient))
+})
+
+app.get('/api/v1/streaming/public', (req, res) => {
+ const redisClient = getRedisClient()
+ streamFrom(redisClient, 'timeline:public', req, streamToHttp(req, res, redisClient), true)
+})
+
+app.get('/api/v1/streaming/hashtag', (req, res) => {
+ const redisClient = getRedisClient()
+ streamFrom(redisClient, `timeline:hashtag:${req.params.tag}`, req, streamToHttp(req, res, redisClient), true)
+})
-log.level = 'verbose'
-log.info(`Starting HTTP server on port ${process.env.PORT || 4000}`)
+wss.on('connection', ws => {
+ const location = url.parse(ws.upgradeReq.url, true)
+ const token = location.query.access_token
+ const req = {}
-app.listen(process.env.PORT || 4000)
+ accountFromToken(token, req, err => {
+ if (err) {
+ log.error(err)
+ ws.close()
+ return
+ }
+
+ const redisClient = getRedisClient()
+
+ switch(location.query.stream) {
+ case 'user':
+ streamFrom(redisClient, `timeline:${req.accountId}`, req, streamToWs(req, ws, redisClient))
+ break;
+ case 'public':
+ streamFrom(redisClient, 'timeline:public', req, streamToWs(req, ws, redisClient), true)
+ break;
+ case 'hashtag':
+ streamFrom(redisClient, `timeline:hashtag:${location.query.tag}`, req, streamToWs(req, ws, redisClient), true)
+ break;
+ default:
+ ws.close()
+ }
+ })
+})
+
+server.listen(process.env.PORT || 4000, () => {
+ log.level = process.env.LOG_LEVEL || 'verbose'
+ log.info(`Starting streaming API server on port ${server.address().port}`)
+})
version "6.15.0"
resolved "https://registry.yarnpkg.com/babylon/-/babylon-6.15.0.tgz#ba65cfa1a80e1759b0e89fb562e27dccae70348e"
+backoff@^2.4.1:
+ version "2.5.0"
+ resolved "https://registry.yarnpkg.com/backoff/-/backoff-2.5.0.tgz#f616eda9d3e4b66b8ca7fca79f695722c5f8e26f"
+ dependencies:
+ precond "0.2"
+
balanced-match@^0.4.1, balanced-match@^0.4.2:
version "0.4.2"
resolved "https://registry.yarnpkg.com/balanced-match/-/balanced-match-0.4.2.tgz#cb3f3e3c732dc0f01ee70b403f302e61d7709838"
version "1.7.0"
resolved "https://registry.yarnpkg.com/binary-extensions/-/binary-extensions-1.7.0.tgz#6c1610db163abfb34edfe42fa423343a1e01185d"
+bindings@~1.2.1:
+ version "1.2.1"
+ resolved "https://registry.yarnpkg.com/bindings/-/bindings-1.2.1.tgz#14ad6113812d2d37d72e67b4cacb4bb726505f11"
+
bl@~1.1.2:
version "1.1.2"
resolved "https://registry.yarnpkg.com/bl/-/bl-1.1.2.tgz#fdca871a99713aa00d19e3bbba41c44787a65398"
ieee754 "^1.1.4"
isarray "^1.0.0"
+bufferutil@^2.0.0:
+ version "2.0.0"
+ resolved "https://registry.yarnpkg.com/bufferutil/-/bufferutil-2.0.0.tgz#6588ed4bafa300798b26dc048494a51abde83507"
+ dependencies:
+ bindings "~1.2.1"
+ nan "~2.5.0"
+
builtin-modules@^1.0.0:
version "1.1.1"
resolved "https://registry.yarnpkg.com/builtin-modules/-/builtin-modules-1.1.1.tgz#270f076c5a72c02f5b65a47df94c5fe3a278892f"
version "0.7.2"
resolved "https://registry.yarnpkg.com/ms/-/ms-0.7.2.tgz#ae25cf2512b3885a1d95d7f037868d8431124765"
-nan@^2.3.0, nan@^2.3.2:
- version "2.4.0"
- resolved "https://registry.yarnpkg.com/nan/-/nan-2.4.0.tgz#fb3c59d45fe4effe215f0b890f8adf6eb32d2232"
+nan@^2.3.0, nan@^2.3.2, nan@~2.5.0:
+ version "2.5.1"
+ resolved "https://registry.yarnpkg.com/nan/-/nan-2.5.1.tgz#d5b01691253326a97a2bbee9e61c55d8d60351e2"
negotiator@0.6.1:
version "0.6.1"
gauge "~2.6.0"
set-blocking "~2.0.0"
-npmlog@4.x, npmlog@^4.0.0:
- version "4.0.0"
- resolved "https://registry.yarnpkg.com/npmlog/-/npmlog-4.0.0.tgz#e094503961c70c1774eb76692080e8d578a9f88f"
- dependencies:
- are-we-there-yet "~1.1.2"
- console-control-strings "~1.1.0"
- gauge "~2.6.0"
- set-blocking "~2.0.0"
-
-npmlog@^4.0.2:
+npmlog@4.x, npmlog@^4.0.0, npmlog@^4.0.2:
version "4.0.2"
resolved "https://registry.yarnpkg.com/npmlog/-/npmlog-4.0.2.tgz#d03950e0e78ce1527ba26d2a7592e9348ac3e75f"
dependencies:
dependencies:
xtend "^4.0.0"
+precond@0.2:
+ version "0.2.3"
+ resolved "https://registry.yarnpkg.com/precond/-/precond-0.2.3.tgz#aa9591bcaa24923f1e0f4849d240f47efc1075ac"
+
prelude-ls@~1.1.2:
version "1.1.2"
resolved "https://registry.yarnpkg.com/prelude-ls/-/prelude-ls-1.1.2.tgz#21932a549f5e52ffd9a827f570e04be62a97da54"
version "0.0.6"
resolved "https://registry.yarnpkg.com/uid-number/-/uid-number-0.0.6.tgz#0ea10e8035e8eb5b8e4449f06da1c730663baa81"
+ultron@~1.1.0:
+ version "1.1.0"
+ resolved "https://registry.yarnpkg.com/ultron/-/ultron-1.1.0.tgz#b07a2e6a541a815fc6a34ccd4533baec307ca864"
+
umd@^3.0.0:
version "3.0.1"
resolved "https://registry.yarnpkg.com/umd/-/umd-3.0.1.tgz#8ae556e11011f63c2596708a8837259f01b3d60e"
version "1.1.1"
resolved "https://registry.yarnpkg.com/user-home/-/user-home-1.1.1.tgz#2b5be23a32b63a7c9deb8d0f28d485724a3df190"
+utf-8-validate@^3.0.0:
+ version "3.0.0"
+ resolved "https://registry.yarnpkg.com/utf-8-validate/-/utf-8-validate-3.0.0.tgz#42e54dfbc7cdfbd1d3bbf0a2f5000b4c6aeaa0c9"
+ dependencies:
+ bindings "~1.2.1"
+ nan "~2.5.0"
+
util-deprecate@~1.0.1:
version "1.0.2"
resolved "https://registry.yarnpkg.com/util-deprecate/-/util-deprecate-1.0.2.tgz#450d4dc9fa70de732762fbd2d4a28981419a0ccf"
watchpack "^0.2.1"
webpack-core "~0.6.9"
+websocket.js@^0.1.7:
+ version "0.1.7"
+ resolved "https://registry.yarnpkg.com/websocket.js/-/websocket.js-0.1.7.tgz#8d24cefb1a080c259e7e4740c02cab8f142df2b0"
+ dependencies:
+ backoff "^2.4.1"
+
whatwg-fetch@>=0.10.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/whatwg-fetch/-/whatwg-fetch-1.0.0.tgz#01c2ac4df40e236aaa18480e3be74bd5c8eb798e"
imurmurhash "^0.1.4"
slide "^1.1.5"
+ws@^2.0.2:
+ version "2.0.2"
+ resolved "https://registry.yarnpkg.com/ws/-/ws-2.0.2.tgz#6257d1a679f0cb23658cba3dcad1316e2b1000c5"
+ dependencies:
+ ultron "~1.1.0"
+
xdg-basedir@^2.0.0:
version "2.0.0"
resolved "https://registry.yarnpkg.com/xdg-basedir/-/xdg-basedir-2.0.0.tgz#edbc903cc385fc04523d966a335504b5504d1bd2"
cat aescling's git repositories / mastodon.git / commitdiff