stations.js 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263
  1. 'use strict';
  2. const async = require('async'),
  3. request = require('request'),
  4. config = require('config');
  5. const io = require('../io');
  6. const db = require('../db');
  7. const cache = require('../cache');
  8. const notifications = require('../notifications');
  9. const utils = require('../utils');
  10. notifications.subscribe('station.locked', function(stationName) {
  11. io.to(`station.${stationName}`).emit("event:station.locked");
  12. });
  13. notifications.subscribe('station.unlocked', function(stationName) {
  14. io.to(`station.${stationName}`).emit("event:station.unlocked");
  15. });
  16. notifications.subscribe('station.pause', function(stationName) {
  17. io.to(`station.${stationName}`).emit("event:station.pause");
  18. });
  19. notifications.subscribe('station.resume', function(stationName) {
  20. io.to(`station.${stationName}`).emit("event:station.resume");
  21. });
  22. /**
  23. * Loads a station into the cache, and sets up all the related logic
  24. *
  25. * @param {String} stationId - the id of the station
  26. * @param {Function} cb - gets called when this function completes
  27. */
  28. function initializeAndReturnStation (stationId, cb) {
  29. async.waterfall([
  30. // first check the cache for the station
  31. (next) => cache.hget('stations', stationId, next),
  32. // if the cached version exist
  33. (station, next) => {
  34. if (station) return next(true, station);
  35. db.models.station.findOne({ id: stationId }, next);
  36. },
  37. // if the station exists in the DB, add it to the cache
  38. (station, next) => {
  39. if (!station) return cb('Station by that id does not exist');
  40. station = cache.schemas.station(station);
  41. cache.hset('stations', station.id, station, (err) => next(err, station));
  42. }
  43. ], (err, station) => {
  44. if (err && err !== true) return cb(err);
  45. // get notified when the next song for this station should play, so that we can notify our sockets
  46. let notification = notifications.subscribe(`stations.nextSong?id=${station.id}`, () => {
  47. // get the station from the cache
  48. console.log('NOTIFICATION');
  49. cache.hget('stations', station.name, (err, station) => {
  50. if (station) {
  51. console.log(777);
  52. // notify all the sockets on this station to go to the next song
  53. io.to(`station.${stationId}`).emit("event:songs.next", {
  54. currentSong: station.currentSong,
  55. startedAt: station.startedAt,
  56. paused: station.paused,
  57. timePaused: 0
  58. });
  59. // schedule a notification to be dispatched when the next song ends
  60. notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
  61. }
  62. // the station doesn't exist anymore, unsubscribe from it
  63. else {
  64. console.log(888);
  65. notifications.remove(notification);
  66. }
  67. });
  68. }, true);
  69. if (!station.paused) {
  70. console.log(station);
  71. notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
  72. }
  73. return cb(null, station);
  74. // will need to be added once station namespace thing is decided
  75. // function generatePlaylist(arr) {
  76. // station.playlist = [];
  77. // return arr.reduce((promise, id) => {
  78. // return promise.then(() => {
  79. // return globals.db.models.song.findOne({ id }, (err, song) => {
  80. // if (err) throw err;
  81. // station.playlist.push(song);
  82. // });
  83. // });
  84. // }, Promise.resolve());
  85. // }
  86. // generatePlaylist(station.playlist).then(() => {
  87. // cb(null, station);
  88. // });
  89. });
  90. }
  91. module.exports = {
  92. /**
  93. * Get a list of all the stations
  94. *
  95. * @param session
  96. * @param cb
  97. * @return {{ status: String, stations: Array }}
  98. */
  99. index: (session, cb) => {
  100. // TODO: the logic should be a bit more personalized to the users preferred genres
  101. // and it should probably just a different cache table then 'stations'
  102. cache.hgetall('stations', (err, stations) => {
  103. if (err && err !== true) {
  104. return cb({
  105. status: 'error',
  106. message: 'An error occurred while obtaining the stations'
  107. });
  108. }
  109. let arr = [];
  110. for (let prop in stations) {
  111. arr.push(stations[prop]);
  112. }
  113. cb({ status: 'success', stations: arr });
  114. });
  115. },
  116. /**
  117. * Joins the station by its id
  118. *
  119. * @param sessionId
  120. * @param stationId - the station id
  121. * @param cb
  122. * @return {{ status: String, userCount: Integer }}
  123. */
  124. join: (sessionId, stationId, cb) => {
  125. initializeAndReturnStation(stationId, (err, station) => {
  126. if (err && err !== true) {
  127. return cb({ status: 'error', message: 'An error occurred while joining the station' });
  128. }
  129. if (station) {
  130. //TODO Loop through all sockets, see if socket with same session exists, and if so leave all other station rooms and join this stationRoom
  131. cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
  132. if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });
  133. utils.socketJoinRoom(sessionId);
  134. //TODO Emit to cache, listen on cache
  135. cb({ status: 'success', currentSong: station.currentSong, startedAt: station.startedAt, paused: station.paused, timePaused: station.timePaused });
  136. });
  137. }
  138. else {
  139. cb({ status: 'failure', message: `That station doesn't exist` });
  140. }
  141. });
  142. },
  143. /**
  144. * Skips the users current station
  145. *
  146. * @param session
  147. * @param stationId - the station id
  148. * @param cb
  149. * @return {{ status: String, skipCount: Integer }}
  150. */
  151. skip: (session, stationId, cb) => {
  152. if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
  153. initializeAndReturnStation(stationId, (err, station) => {
  154. if (err && err !== true) {
  155. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  156. }
  157. if (station) {
  158. cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
  159. session.skippedSong = true;
  160. if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  161. cache.hget('station.userCounts', session.stationId, (err, userCount) => {
  162. cb({ status: 'success', skipCount });
  163. });
  164. });
  165. }
  166. else {
  167. cb({ status: 'failure', message: `That station doesn't exist` });
  168. }
  169. });
  170. },
  171. /**
  172. * Leaves the users current station
  173. *
  174. * @param session
  175. * @param cb
  176. * @return {{ status: String, userCount: Integer }}
  177. */
  178. leave: (session, cb) => {
  179. let stationId = "edm";
  180. initializeAndReturnStation(stationId, (err, station) => {
  181. if (err && err !== true) {
  182. return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  183. }
  184. if (session) session.stationId = null;
  185. else if (station) {
  186. cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
  187. if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  188. utils.socketLeaveRooms(session);
  189. cb({ status: 'success', userCount });
  190. });
  191. } else {
  192. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  193. }
  194. });
  195. },
  196. lock: (session, stationId, cb) => {
  197. //TODO Require admin
  198. initializeAndReturnStation(stationId, (err, station) => {
  199. if (err && err !== true) {
  200. return cb({ status: 'error', message: 'An error occurred while locking the station' });
  201. } else if (station) {
  202. // Add code to update Mongo and Redis
  203. cb({ status: 'success' });
  204. } else {
  205. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  206. }
  207. });
  208. },
  209. unlock: (session, stationId, cb) => {
  210. //TODO Require admin
  211. initializeAndReturnStation(stationId, (err, station) => {
  212. if (err && err !== true) {
  213. return cb({ status: 'error', message: 'An error occurred while unlocking the station' });
  214. } else if (station) {
  215. // Add code to update Mongo and Redis
  216. cb({ status: 'success' });
  217. } else {
  218. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  219. }
  220. });
  221. },
  222. };