Files
Iris/src/js/services/pusher/middleware.js
2020-03-03 16:39:37 +13:00

819 lines
25 KiB
JavaScript
Executable File

import ReactGA from 'react-ga';
import { uriType, generateGuid } from '../../util/helpers';
const coreActions = require('../core/actions');
const uiActions = require('../ui/actions');
const pusherActions = require('./actions');
const lastfmActions = require('../lastfm/actions');
const geniusActions = require('../genius/actions');
const spotifyActions = require('../spotify/actions');
const PusherMiddleware = (function () {
// container for the actual websocket
let socket = null;
let reconnectTimer = null;
// requests pending
const deferredRequests = [];
// handle all manner of socket messages
const handleMessage = (ws, store, message) => {
if (store.getState().ui.log_pusher) {
console.log('Pusher log (incoming)', message);
}
// Pull our ID. JSON-RPC nests the ID under the error object,
// so make sure we handle that.
// TODO: Use this as our measure of a successful response vs error
let id = null;
if (message.id) {
id = message.id;
} else if (message.error && message.error.id) {
id = message.error.id;
}
// Response with request_id
if (id) {
// Response matches a pending request
if (deferredRequests[id] !== undefined) {
store.dispatch(uiActions.stopLoading(id));
// Response is an error
if (message.error !== undefined) {
deferredRequests[id].reject(message.error);
// Successful response
} else {
deferredRequests[id].resolve(message.result);
}
// Hmm, the response doesn't appear to be for us?
} else {
store.dispatch(coreActions.handleException(
'Pusher: Response received with no matching request',
message,
));
}
// Broadcast of an error
} else if (message.error !== undefined) {
store.dispatch(coreActions.handleException(
`Pusher: ${message.error.message}`,
message,
(message.error.data !== undefined && message.error.data.description !== undefined
? message.error.data.description
: null
),
));
// General broadcast received
} else {
const params = message.params ? message.params : {};
switch (message.method) {
case 'connection_added':
store.dispatch(pusherActions.connectionAdded(params.connection));
break;
case 'connection_changed':
store.dispatch(pusherActions.connectionChanged(params.connection));
break;
case 'connection_removed':
store.dispatch(pusherActions.connectionRemoved(params.connection));
break;
case 'queue_metadata_changed':
store.dispatch(pusherActions.queueMetadataChanged(params.queue_metadata));
break;
case 'spotify_token_changed':
store.dispatch(spotifyActions.tokenChanged(params.spotify_token));
break;
case 'share_configuration_received':
store.dispatch(uiActions.createNotification({
type: 'share-configuration-received',
configuration: params,
sticky: true,
}));
break;
case 'notification':
store.dispatch(uiActions.createNotification(params.notification));
break;
case 'radio_started':
store.dispatch(pusherActions.radioStarted(params.radio));
break;
case 'radio_changed':
store.dispatch(pusherActions.radioChanged(params.radio));
break;
case 'radio_stopped':
store.dispatch(pusherActions.radioStopped());
break;
case 'commands_changed':
store.dispatch(pusherActions.commandsUpdated(params.commands));
break;
case 'reload':
window.location.reload(true);
break;
// Local scan
case 'local_scan_started':
store.dispatch(uiActions.updateProcess('local_scan', 'Scanning local library'));
break;
case 'local_scan_updated':
store.dispatch(uiActions.updateProcess('local_scan', 'Scanning local library', {}, params.output));
break;
case 'local_scan_finished':
store.dispatch(uiActions.processFinished(
'local_scan',
{
content: 'Local scan finished', description: params.output, sticky: true,
},
));
break;
case 'local_scan_error':
store.dispatch(uiActions.processFinished(
'local_scan',
{
level: 'error', content: 'Local scan failed', description: params.error, sticky: true,
},
));
break;
// Upgrade
case 'upgrade_started':
store.dispatch(uiActions.updateProcess('upgrade', 'Upgrading'));
break;
case 'upgrade_updated':
store.dispatch(uiActions.updateProcess('upgrade', 'Upgrading', {}, params.output));
break;
case 'upgrade_finished':
store.dispatch(uiActions.updateProcess('upgrade', 'Restarting to complete upgrade'));
break;
case 'upgrade_error':
store.dispatch(uiActions.processFinished(
'upgrade',
{
level: 'error', content: 'Upgrade failed', description: params.error, sticky: true,
},
));
break;
// Restart
case 'restart_started':
store.dispatch(uiActions.removeProcess('upgrade', 'Restarting'));
break;
case 'restart_updated':
store.dispatch(uiActions.updateProcess('upgrade', 'Restarting', {}, params.output));
break;
case 'restart_error':
store.dispatch(uiActions.processFinished(
'upgrade',
{
level: 'error', content: 'Restart failed', description: params.error, sticky: true,
},
));
break;
// Test
case 'test_started':
store.dispatch(uiActions.updateProcess('test', 'Running test', {}, params.output));
break;
case 'test_updated':
store.dispatch(uiActions.updateProcess('test', 'Running test'));
break;
case 'test_finished':
store.dispatch(uiActions.processFinished(
'test',
{
content: 'Test finished', description: params.output, sticky: true,
},
));
break;
case 'test_error':
store.dispatch(uiActions.removeProcess(
'test',
{
level: 'error', content: params.message, description: params.error,
},
));
break;
default:
break;
}
}
};
const request = (store, method, params = null) => new Promise((resolve, reject) => {
const id = generateGuid();
const message = {
jsonrpc: '2.0',
id,
method,
};
if (params) {
message.params = params;
}
if (store.getState().ui.log_pusher) {
console.log('Pusher log (outgoing)', message);
}
socket.send(JSON.stringify(message));
store.dispatch(uiActions.startLoading(id, `pusher_${method}`));
// Start our 30 second timeout
setTimeout(
() => {
store.dispatch(uiActions.stopLoading(id));
reject({
id,
code: 32300,
message: 'Request timed out',
});
},
30000,
);
// add query to our deferred responses
deferredRequests[id] = {
resolve,
reject,
};
});
return (store) => (next) => (action) => {
const { pusher } = store.getState();
switch (action.type) {
case 'PUSHER_CONNECT':
if (socket != null) {
socket.close();
}
clearTimeout(reconnectTimer);
store.dispatch({ type: 'PUSHER_CONNECTING' });
socket = new WebSocket(
`ws${window.location.protocol === 'https:' ? 's' : ''}://${store.getState().mopidy.host}:${store.getState().mopidy.port}/iris/ws/`,
);
socket.onopen = () => {
store.dispatch({
type: 'PUSHER_CONNECTED',
});
};
socket.onclose = (e) => {
store.dispatch({
type: 'PUSHER_DISCONNECTED',
});
// attempt to reconnect every 5 seconds
reconnectTimer = setTimeout(() => {
store.dispatch(pusherActions.connect());
}, 5000);
};
socket.onerror = (e) => {
if (socket.readyState == 1) {
store.dispatch(coreActions.handleException(
'Pusher websocket error',
e,
e.type,
));
}
};
socket.onmessage = (message) => {
handleMessage(socket, store, JSON.parse(message.data));
};
break;
case 'PUSHER_CONNECTED':
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Connected', label: action.username });
}
clearTimeout(reconnectTimer);
store.dispatch(pusherActions.updateConnection());
store.dispatch(pusherActions.getConfig());
store.dispatch(pusherActions.getRadio());
store.dispatch(pusherActions.getCommands());
store.dispatch(pusherActions.getQueueMetadata());
// Give things a few moments to setup before we check for version.
// This is because the server makes a GitHub request, which creates a [very] small delay
// in subsequent requests.
setTimeout(
() => {
store.dispatch(pusherActions.getVersion());
},
500,
);
next(action);
break;
case 'PUSHER_REQUEST':
request(store, action.method, action.params)
.then(
(response) => {
if (action.response_callback) {
action.response_callback.call(this, response);
}
},
(error) => {
if (action.error_callback) {
action.error_callback.call(this, error);
} else {
store.dispatch(coreActions.handleException(
'Pusher request failed',
error,
action.method,
action,
));
}
},
);
break;
case 'PUSHER_DELIVER_MESSAGE':
request(store, 'send_message', action.data)
.then(
(response) => {
store.dispatch(uiActions.createNotification({ level: 'warning', content: 'Message delivered' }));
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not deliver message',
error,
));
},
);
break;
case 'PUSHER_DELIVER_BROADCAST':
request(store, 'broadcast', action.data);
break;
case 'PUSHER_SET_USERNAME':
store.dispatch(pusherActions.updateConnection({ username: action.username }));
next(action);
break;
case 'PUSHER_UPDATE_CONNECTION':
// Our action can provide new values during a state update (eg the field was just changed)
// but by default we refer to our existing state
const connection = {
username: store.getState().pusher.username,
client_id: store.getState().pusher.client_id,
...(action.connection ? action.connection : {}),
};
request(store, 'update_connection', connection)
.then(
(response) => {
response.type = 'PUSHER_CONNECTION_UPDATED';
store.dispatch(response);
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not update connection',
error,
));
},
);
next(action);
break;
case 'PUSHER_GET_QUEUE_METADATA':
request(store, 'get_queue_metadata')
.then(
(response) => {
response.type = 'PUSHER_QUEUE_METADATA';
store.dispatch(response);
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not load queue metadata',
error,
));
},
);
break;
case 'PUSHER_ADD_QUEUE_METADATA':
request(store, 'add_queue_metadata', {
tlids: action.tlids,
added_from: action.from_uri,
added_by: pusher.username,
});
break;
case 'PUSHER_GET_VERSION':
request(store, 'get_version')
.then(
(response) => {
store.dispatch({
type: 'PUSHER_VERSION',
version: response.version,
});
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not load version',
error,
));
},
);
break;
case 'PUSHER_GET_CONFIG':
request(store, 'get_config')
.then(
(response) => {
store.dispatch({
type: 'PUSHER_CONFIG',
config: response.config,
});
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not load config',
error,
));
},
);
break;
case 'PUSHER_GET_CONNECTIONS':
request(store, 'get_connections')
.then(
(response) => {
store.dispatch({
type: 'PUSHER_CONNECTIONS',
connections: response.connections,
});
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not load connections',
error,
));
},
);
return next(action);
break;
/**
* Commands
* */
case 'PUSHER_GET_COMMANDS':
request(store, 'get_commands')
.then(
(response) => {
store.dispatch(pusherActions.commandsUpdated(response.commands));
},
(error) => {
// We're not too worried about capturing errors here
// It's also likely to fail where UI has been updated but
// server hasn't been restarted yet.
},
);
next(action);
break;
case 'PUSHER_SET_COMMAND':
var commands = { ...pusher.commands };
if (commands[action.command.id]) {
var command = { ...commands[action.command.id], ...action.command };
} else {
var { command } = action;
}
commands[action.command.id] = command;
store.dispatch(pusherActions.setCommands(commands));
break;
case 'PUSHER_SET_COMMANDS':
request(store, 'set_commands', { commands: action.commands })
.then(
(response) => {
// No action required, the change will be broadcast
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not set commands',
error,
));
},
);
next(action);
break;
case 'PUSHER_REMOVE_COMMAND':
var commands_index = { ...pusher.commands };
delete commands_index[action.id];
request(store, 'set_commands', { commands: commands_index })
.then(
(response) => {
// No action required, the change will be broadcast
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not remove command',
error,
));
},
);
next(action);
break;
case 'PUSHER_RUN_COMMAND':
var command = { ...pusher.commands[action.id] };
var notification_key = `command_${action.id}`;
if (action.notify) {
store.dispatch(uiActions.startProcess(notification_key, 'Running command'));
}
request(store, 'run_command', { id: action.id })
.then((response) => {
console.log('Command response', response);
store.dispatch(uiActions.removeProcess(notification_key));
if (action.notify) {
store.dispatch(uiActions.createNotification({ key: notification_key, level: 'warning', content: 'Command sent' }));
}
},
(error) => {
store.dispatch(uiActions.removeProcess(notification_key));
store.dispatch(coreActions.handleException(
'Could not run command',
error,
));
});
break;
/**
* Radio
* */
case 'PUSHER_GET_RADIO':
request(store, 'get_radio')
.then(
(response) => {
store.dispatch({
type: 'PUSHER_RADIO_LOADED',
radio: response.radio,
});
if (response.radio.enabled) {
store.dispatch(spotifyActions.resolveRadioSeeds(response.radio));
}
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not load radio',
error,
));
},
);
break;
case 'PUSHER_START_RADIO':
case 'PUSHER_UPDATE_RADIO':
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Start radio', label: action.uris.join() });
}
// start our UI process notification
if (action.type == 'PUSHER_UPDATE_RADIO') {
store.dispatch(uiActions.startProcess('PUSHER_RADIO_PROCESS', 'Updating radio'));
} else {
store.dispatch(uiActions.startProcess('PUSHER_RADIO_PROCESS', 'Starting radio'));
}
var data = {
reset: (action.type == 'PUSHER_START_RADIO'),
seed_artists: [],
seed_genres: [],
seed_tracks: [],
};
for (let i = 0; i < action.uris.length; i++) {
switch (uriType(action.uris[i])) {
case 'artist':
data.seed_artists.push(action.uris[i]);
break;
case 'track':
data.seed_tracks.push(action.uris[i]);
break;
case 'genre':
data.seed_genres.push(action.uris[i]);
break;
}
}
if (action.type == 'PUSHER_START_RADIO') {
store.dispatch(pusherActions.deliverBroadcast(
'notification',
{
notification: {
level: 'warning',
content: `${pusher.username} is starting radio mode`,
},
},
));
}
request(store, 'change_radio', data)
.then(
(response) => {
store.dispatch(uiActions.processFinished('PUSHER_RADIO_PROCESS'));
if (response.status == 0) {
store.dispatch(uiActions.createNotification({ content: response.message, level: 'error' }));
}
store.dispatch(pusherActions.radioChanged(response.radio));
},
(error) => {
store.dispatch(uiActions.processFinished('PUSHER_RADIO_PROCESS'));
store.dispatch(coreActions.handleException(
'Could not change radio',
error,
));
},
);
break;
case 'PUSHER_STOP_RADIO':
store.dispatch(uiActions.createNotification({ content: 'Stopping radio' }));
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Stop radio' });
}
store.dispatch(pusherActions.deliverBroadcast(
'notification',
{
notification: {
level: 'warning',
content: `${pusher.username} stopped radio mode`,
},
},
));
var data = {
seed_artists: [],
seed_genres: [],
seed_tracks: [],
};
request(store, 'stop_radio', data)
.then(
(response) => {
store.dispatch(pusherActions.radioStopped());
}, (error) => {
store.dispatch(coreActions.handleException(
'Could not stop radio',
error,
));
},
);
break;
/**
* Notifications and alerts
* */
case 'PUSHER_BROWSER_NOTIFICATION':
store.dispatch(uiActions.createBrowserNotification(action));
break;
case 'PUSHER_NOTIFICATION':
var data = {
...action, type: action.notification_type,
};
store.dispatch(uiActions.createNotification(data));
break;
/**
* Server actions
* */
case 'PUSHER_RELOAD':
// Hard reload. This doesn't strictly clear the cache, but our compiler's
// cache buster should handle that
window.location.reload(true);
break;
case 'PUSHER_RESTART':
request(store, 'restart');
next(action);
break;
case 'PUSHER_UPGRADE':
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Upgrade', label: '' });
}
request(store, 'upgrade');
break;
case 'PUSHER_LOCAL_SCAN':
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Local scan', label: '' });
}
request(store, 'local_scan');
break;
case 'PUSHER_TEST':
request(store, 'test');
break;
case 'PUSHER_VERSION':
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Version', label: action.version.current });
}
next(action);
break;
case 'PUSHER_CONFIG':
// Set default country/locale (unless we've already been configured)
var { spotify } = store.getState();
var spotify_updated = false;
var spotify_updates = {};
if (!spotify.country && action.config.country) {
spotify_updates.country = action.config.country;
spotify_updated = true;
}
if (!spotify.locale && action.config.locale) {
spotify_updates.locale = action.config.locale;
spotify_updated = true;
}
if (action.config.spotify_authorization_url) {
spotify_updates.authorization_url = action.config.spotify_authorization_url;
spotify_updated = true;
}
if (spotify_updated) {
store.dispatch(spotifyActions.set(spotify_updates));
}
store.dispatch(lastfmActions.set({
authorization_url: (action.config.lastfm_authorization_url ? action.config.lastfm_authorization_url : null),
}));
store.dispatch(geniusActions.set({
authorization_url: (action.config.genius_authorization_url ? action.config.genius_authorization_url : null),
}));
next(action);
break;
case 'PUSHER_DEBUG':
request(store, action.message.method, action.message.data)
.then(
(response) => {
store.dispatch({ type: 'DEBUG', response });
},
(error) => {
store.dispatch(coreActions.handleException(
'Could not debug',
error,
error.message,
));
},
);
break;
case 'PUSHER_ERROR':
store.dispatch(uiActions.createNotification(action.message, 'bad'));
if (store.getState().ui.allow_reporting) {
ReactGA.event({ category: 'Pusher', action: 'Error', label: action.message });
}
break;
// This action is irrelevant to us, pass it on to the next middleware
default:
return next(action);
}
};
}());
export default PusherMiddleware;