123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278 |
- 'use strict';
- const async = require('async'),
- request = require('request'),
- config = require('config');
- const io = require('../io');
- const db = require('../db');
- const cache = require('../cache');
- const notifications = require('../notifications');
- const utils = require('../utils');
- let stationsLoaded = {};
- notifications.subscribe('station.locked', function(stationName) {
- io.to(`station.${stationName}`).emit("event:station.locked");
- });
- notifications.subscribe('station.unlocked', function(stationName) {
- io.to(`station.${stationName}`).emit("event:station.unlocked");
- });
- notifications.subscribe('station.pause', function(stationName) {
- io.to(`station.${stationName}`).emit("event:station.pause");
- });
- notifications.subscribe('station.resume', function(stationName) {
- io.to(`station.${stationName}`).emit("event:station.resume");
- });
- /**
- * Loads a station into the cache, and sets up all the related logic
- *
- * @param {String} stationId - the id of the station
- * @param {Function} cb - gets called when this function completes
- */
- function initializeAndReturnStation (stationId, cb) {
- async.waterfall([
- // first check the cache for the station
- (next) => cache.hget('stations', stationId, next),
- // if the cached version exist
- (station, next) => {
- if (station) return next(true, station);
- db.models.station.findOne({ id: stationId }, next);
- },
- // if the station exists in the DB, add it to the cache
- (station, next) => {
- if (!station) return cb('Station by that id does not exist');
- station = cache.schemas.station(station);
- cache.hset('stations', station.id, station, (err) => next(err, station));
- }
- ], (err, station) => {
- if (err && err !== true) return cb(err);
- // get notified when the next song for this station should play, so that we can notify our sockets
- if (stationsLoaded[stationId] === undefined) {
- stationsLoaded[stationId] = 1;
- let notification = notifications.subscribe(`stations.nextSong?id=${station.id}`, () => {
- // get the station from the cache
- cache.hget('stations', station.name, (err, station) => {
- if (station) {
- console.log(777);
- // notify all the sockets on this station to go to the next song
- io.to(`station.${stationId}`).emit("event:songs.next", {
- currentSong: station.currentSong,
- startedAt: station.startedAt,
- paused: station.paused,
- timePaused: 0
- });
- // schedule a notification to be dispatched when the next song ends
- notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
- }
- // the station doesn't exist anymore, unsubscribe from it
- else {
- console.log(888);
- notifications.remove(notification);
- delete stationsLoaded[stationId];
- }
- });
- }, true);
- if (!station.paused) {
- console.log(station);
- notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
- }
- }
- return cb(null, station);
- // will need to be added once station namespace thing is decided
- // function generatePlaylist(arr) {
- // station.playlist = [];
- // return arr.reduce((promise, id) => {
- // return promise.then(() => {
- // return globals.db.models.song.findOne({ id }, (err, song) => {
- // if (err) throw err;
- // station.playlist.push(song);
- // });
- // });
- // }, Promise.resolve());
- // }
- // generatePlaylist(station.playlist).then(() => {
- // cb(null, station);
- // });
- });
- }
- module.exports = {
- /**
- * Get a list of all the stations
- *
- * @param session
- * @param cb
- * @return {{ status: String, stations: Array }}
- */
- index: (session, cb) => {
- // TODO: the logic should be a bit more personalized to the users preferred genres
- // and it should probably just a different cache table then 'stations'
- cache.hgetall('stations', (err, stations) => {
- if (err && err !== true) {
- return cb({
- status: 'error',
- message: 'An error occurred while obtaining the stations'
- });
- }
- let arr = [];
- for (let prop in stations) {
- arr.push(stations[prop]);
- }
- cb({ status: 'success', stations: arr });
- });
- },
- /**
- * Joins the station by its id
- *
- * @param session
- * @param stationId - the station id
- * @param cb
- * @return {{ status: String, userCount: Integer }}
- */
- join: (session, stationId, cb) => {
- let ns = io.io.of("/");
- if (ns) {
- for (let id in ns.connected) {
- console.log(ns.connected[id]);
- console.log(ns.connected[id].testProp);
- }
- }
- initializeAndReturnStation(stationId, (err, station) => {
- if (err && err !== true) {
- return cb({ status: 'error', message: 'An error occurred while joining the station' });
- }
- if (station) {
- if (session) session.stationId = stationId;
- //TODO Loop through all sockets, see if socket with same session exists, and if so leave all other station rooms and join this stationRoom
- cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
- if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });
- utils.socketJoinRoom(session);
- //TODO Emit to cache, listen on cache
- cb({ status: 'success', currentSong: station.currentSong, startedAt: station.startedAt, paused: station.paused, timePaused: station.timePaused });
- });
- }
- else {
- cb({ status: 'failure', message: `That station doesn't exist` });
- }
- });
- },
- /**
- * Skips the users current station
- *
- * @param session
- * @param stationId - the station id
- * @param cb
- * @return {{ status: String, skipCount: Integer }}
- */
- skip: (session, stationId, cb) => {
- if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
- initializeAndReturnStation(stationId, (err, station) => {
- if (err && err !== true) {
- return cb({ status: 'error', message: 'An error occurred while skipping the station' });
- }
- if (station) {
- cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
- session.skippedSong = true;
- if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
- cache.hget('station.userCounts', session.stationId, (err, userCount) => {
- cb({ status: 'success', skipCount });
- });
- });
- }
- else {
- cb({ status: 'failure', message: `That station doesn't exist` });
- }
- });
- },
- /**
- * Leaves the users current station
- *
- * @param session
- * @param cb
- * @return {{ status: String, userCount: Integer }}
- */
- leave: (session, cb) => {
- let stationId = "edm";
- initializeAndReturnStation(stationId, (err, station) => {
- if (err && err !== true) {
- return cb({ status: 'error', message: 'An error occurred while leaving the station' });
- }
- if (session) session.stationId = null;
- else if (station) {
- cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
- if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
- utils.socketLeaveRooms(session);
- cb({ status: 'success', userCount });
- });
- } else {
- cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
- }
- });
- },
- lock: (session, stationId, cb) => {
- //TODO Require admin
- initializeAndReturnStation(stationId, (err, station) => {
- if (err && err !== true) {
- return cb({ status: 'error', message: 'An error occurred while locking the station' });
- } else if (station) {
- // Add code to update Mongo and Redis
- cb({ status: 'success' });
- } else {
- cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
- }
- });
- },
- unlock: (session, stationId, cb) => {
- //TODO Require admin
- initializeAndReturnStation(stationId, (err, station) => {
- if (err && err !== true) {
- return cb({ status: 'error', message: 'An error occurred while unlocking the station' });
- } else if (station) {
- // Add code to update Mongo and Redis
- cb({ status: 'success' });
- } else {
- cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
- }
- });
- },
- };
|