Move peer stuff to actions

src/client/__tests__/App-test.js

@@ -46,11 +46,8 @@ describe('App', () => {
   describe('state', () => {
     let alert
     beforeEach(() => {
-      state.streams = state.streams.setIn(['all'], {
+      state.streams = state.streams.merge({
-        'test': {
+        test: 'blob://'
-          userId: 'test',
-          url: 'blob://'
-        }
       })
       state.notifications = state.notifications.merge({
         'notification1': {
@@ -87,7 +84,7 @@ describe('App', () => {
         const video = node.querySelector('video')
         TestUtils.Simulate.click(video)
         expect(store.getActions()).toEqual([{
-          type: constants.STREAM_ACTIVATE,
+          type: constants.ACTIVE_SET,
           payload: { userId: 'test' }
         }])
       })

src/client/actions/CallActions.js

@@ -1,10 +1,10 @@
-import * as StreamActions from './StreamActions.js'
 import * as NotifyActions from './NotifyActions.js'
+import * as SocketActions from './SocketActions.js'
+import * as StreamActions from './StreamActions.js'
 import * as constants from '../constants.js'
 import Promise from 'bluebird'
 import callId from '../callId.js'
 import getUserMedia from '../window/getUserMedia.js'
-import handshake from '../peer/handshake.js'
 import socket from '../socket.js'
 
 export const init = () => dispatch => {
@@ -15,7 +15,11 @@ export const init = () => dispatch => {
       getCameraStream()(dispatch)
     ])
     .spread((socket, stream) => {
-      handshake({ socket, callId, stream })
+      dispatch(SocketActions.handshake({
+        socket,
+        roomName: callId,
+        stream
+      }))
     })
   })
 }

src/client/actions/PeerActions.js

@@ -0,0 +1,129 @@
+import * as NotifyActions from '../actions/NotifyActions.js'
+import * as StreamActions from '../actions/StreamActions.js'
+import * as constants from '../constants.js'
+import Peer from 'simple-peer'
+import _ from 'underscore'
+import _debug from 'debug'
+import iceServers from '../iceServers.js'
+import { play } from '../window/video.js'
+
+const debug = _debug('peercalls')
+
+class PeerHandler {
+  constructor ({ socket, user, stream, dispatch, getState }) {
+    this.socket = socket
+    this.user = user
+    this.stream = stream
+    this.dispatch = dispatch
+    this.getState = getState
+  }
+  handleError = err => {
+    const { dispatch, getState, user } = this
+    debug('peer: %s, error %s', user.id, err.stack)
+    dispatch(NotifyActions.error('A peer connection error occurred'))
+    const peer = getState().peers[user.id]
+    peer && peer.destroy()
+    dispatch(removePeer(user.id))
+  }
+  handleSignal = signal => {
+    const { socket, user } = this
+    debug('peer: %s, signal: %o', user.id, signal)
+
+    const payload = { userId: user.id, signal }
+    socket.emit('signal', payload)
+  }
+  handleConnect = () => {
+    const { dispatch, user } = this
+    debug('peer: %s, connect', user.id)
+    dispatch(NotifyActions.warning('Peer connection established'))
+    play()
+  }
+  handleStream = stream => {
+    const { user, dispatch } = this
+    debug('peer: %s, stream', user.id)
+    dispatch(StreamActions.addStream({
+      userId: user.id,
+      stream
+    }))
+  }
+  handleData = object => {
+    const { dispatch, user } = this
+    object = JSON.parse(new window.TextDecoder('utf-8').decode(object))
+    debug('peer: %s, message: %o', user.id, object)
+    const message = user.id + ': ' + object.message
+    dispatch(NotifyActions.info(message))
+  }
+  handleClose = () => {
+    const { dispatch, user } = this
+    debug('peer: %s, close', user.id)
+    dispatch(NotifyActions.error('Peer connection closed'))
+    dispatch(StreamActions.removeStream(user.id))
+    dispatch(removePeer(user.id))
+  }
+}
+
+/**
+ * @param {Object} options
+ * @param {Socket} options.socket
+ * @param {User} options.user
+ * @param {String} options.user.id
+ * @param {Boolean} [options.initiator=false]
+ * @param {MediaStream} [options.stream]
+ */
+export function createPeer ({ socket, user, initiator, stream }) {
+  return (dispatch, getState) => {
+    const userId = user.id
+    debug('create peer: %s, stream:', userId, stream)
+    dispatch(NotifyActions.warning('Connecting to peer...'))
+
+    const oldPeer = getState().peers[userId]
+    if (oldPeer) {
+      dispatch(NotifyActions.info('Cleaning up old connection...'))
+      oldPeer.destroy()
+      dispatch(removePeer(userId))
+    }
+
+    const peer = new Peer({
+      initiator: socket.id === initiator,
+      stream,
+      config: { iceServers }
+    })
+
+    const handler = new PeerHandler({
+      socket,
+      user,
+      stream,
+      dispatch,
+      getState
+    })
+
+    peer.once(constants.PEER_EVENT_ERROR, handler.handleError)
+    peer.once(constants.PEER_EVENT_CONNECT, handler.handleConnect)
+    peer.once(constants.PEER_EVENT_CLOSE, handler.handleClose)
+    peer.on(constants.PEER_EVENT_SIGNAL, handler.handleSignal)
+    peer.on(constants.PEER_EVENT_STREAM, handler.handleStream)
+    peer.on(constants.PEER_EVENT_DATA, handler.handleData)
+
+    dispatch(addPeer({ peer, userId }))
+  }
+}
+
+export const addPeer = ({ peer, userId }) => ({
+  type: constants.PEER_ADD,
+  payload: { peer, userId }
+})
+
+export const removePeer = userId => ({
+  type: constants.PEER_REMOVE,
+  payload: { userId }
+})
+
+export const destroyPeers = () => ({
+  type: constants.PEERS_DESTROY
+})
+
+export const sendMessage = message => (dispatch, getState) => {
+  message = JSON.stringify({ message })
+  const { peers } = getState()
+  _.each(peers, peer => peer.send(message))
+}

src/client/actions/SocketActions.js

@@ -0,0 +1,64 @@
+import * as NotifyActions from '../actions/NotifyActions.js'
+import * as PeerActions from '../actions/PeerActions.js'
+import * as constants from '../constants.js'
+import _ from 'underscore'
+import _debug from 'debug'
+
+const debug = _debug('peercalls')
+
+class SocketHandler {
+  constructor ({ socket, roomName, stream, dispatch, getState }) {
+    this.socket = socket
+    this.roomName = roomName
+    this.stream = stream
+    this.dispatch = dispatch
+    this.getState = getState
+  }
+  handleSignal = ({ userId, signal }) => {
+    const { getState } = this
+    const peer = getState().peers[userId]
+    // debug('socket signal, userId: %s, signal: %o', userId, signal);
+    if (!peer) return debug('user: %s, no peer found', userId)
+    peer.signal(signal)
+  }
+  handleUsers = ({ initiator, users }) => {
+    const { socket, stream, dispatch, getState } = this
+    debug('socket users: %o', users)
+    dispatch(NotifyActions.info('Connected users: {0}', users.length))
+    const { peers } = getState()
+
+    users
+    .filter(user => !peers[user.id] && user.id !== socket.id)
+    .forEach(user => dispatch(PeerActions.createPeer({
+      socket,
+      user,
+      initiator,
+      stream
+    })))
+
+    let newUsersMap = _.indexBy(users, 'id')
+    _.keys(peers)
+    .filter(id => !newUsersMap[id])
+    .forEach(id => peers[id].destroy())
+  }
+}
+
+export function handshake ({ socket, roomName, stream }) {
+  return (dispatch, getState) => {
+    const handler = new SocketHandler({
+      socket,
+      roomName,
+      stream,
+      dispatch,
+      getState
+    })
+
+    socket.on(constants.SOCKET_EVENT_SIGNAL, handler.handleSignal)
+    socket.on(constants.SOCKET_EVENT_USERS, handler.handleUsers)
+
+    debug('socket.id: %s', socket.id)
+    debug('emit ready for room: %s', roomName)
+    dispatch(NotifyActions.info('Ready for connections'))
+    socket.emit('ready', roomName)
+  }
+}

src/client/actions/StreamActions.js

@@ -13,7 +13,7 @@ export const removeStream = userId => ({
   payload: { userId }
 })
 
-export const activateStream = userId => ({
+export const setActive = userId => ({
-  type: constants.STREAM_ACTIVATE,
+  type: constants.ACTIVE_SET,
   payload: { userId }
 })

src/client/actions/__tests__/CallActions-test.js

@@ -1,15 +1,15 @@
 jest.mock('../../callId.js')
 jest.mock('../../iceServers.js')
-jest.mock('../../peer/handshake.js')
 jest.mock('../../socket.js')
 jest.mock('../../window/getUserMedia.js')
 jest.mock('../../store.js')
+jest.mock('../SocketActions.js')
 
 import * as CallActions from '../CallActions.js'
+import * as SocketActions from '../SocketActions.js'
 import * as constants from '../../constants.js'
 import * as getUserMediaMock from '../../window/getUserMedia.js'
 import callId from '../../callId.js'
-import handshake from '../../peer/handshake.js'
 import socket from '../../socket.js'
 import store from '../../store.js'
 
@@ -20,6 +20,7 @@ describe('reducers/alerts', () => {
   beforeEach(() => {
     store.clearActions()
     getUserMediaMock.fail(false)
+    SocketActions.handshake.mockReturnValue(jest.fn())
   })
 
   afterEach(() => {
@@ -43,9 +44,9 @@ describe('reducers/alerts', () => {
         }
       }])
       promise.then(() => {
-        expect(handshake.mock.calls).toEqual([[{
+        expect(SocketActions.handshake.mock.calls).toEqual([[{
           socket,
-          callId,
+          roomName: callId,
           stream: getUserMediaMock.stream
         }]])
       })

src/client/actions/__tests__/PeerActions-test.js

@@ -0,0 +1,172 @@
+jest.mock('../../window/video.js')
+jest.mock('../../callId.js')
+jest.mock('../../iceServers.js')
+jest.mock('simple-peer')
+
+import * as PeerActions from '../PeerActions.js'
+import Peer from 'simple-peer'
+import { EventEmitter } from 'events'
+import { createStore } from '../../store.js'
+import { play } from '../../window/video.js'
+
+describe('PeerActions', () => {
+  function createSocket () {
+    const socket = new EventEmitter()
+    socket.id = 'user1'
+    return socket
+  }
+
+  let socket, stream, user, store
+  beforeEach(() => {
+    store = createStore()
+
+    user = { id: 'user2' }
+    socket = createSocket()
+    Peer.instances = []
+    Peer.mockClear()
+    play.mockClear()
+    stream = { stream: true }
+  })
+
+  describe('create', () => {
+    it('creates a new peer', () => {
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user2', stream })
+      )
+
+      expect(Peer.instances.length).toBe(1)
+      expect(Peer.mock.calls.length).toBe(1)
+      expect(Peer.mock.calls[0][0].initiator).toBe(false)
+      expect(Peer.mock.calls[0][0].stream).toBe(stream)
+    })
+
+    it('sets initiator correctly', () => {
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user1', stream })
+      )
+
+      expect(Peer.instances.length).toBe(1)
+      expect(Peer.mock.calls.length).toBe(1)
+      expect(Peer.mock.calls[0][0].initiator).toBe(true)
+      expect(Peer.mock.calls[0][0].stream).toBe(stream)
+    })
+
+    it('destroys old peer before creating new one', () => {
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user2', stream })
+      )
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user2', stream })
+      )
+
+      expect(Peer.instances.length).toBe(2)
+      expect(Peer.mock.calls.length).toBe(2)
+      expect(Peer.instances[0].destroy.mock.calls.length).toBe(1)
+      expect(Peer.instances[1].destroy.mock.calls.length).toBe(0)
+    })
+  })
+
+  describe('events', () => {
+    let peer
+
+    beforeEach(() => {
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user1', stream })
+      )
+      peer = Peer.instances[0]
+    })
+
+    describe('connect', () => {
+      beforeEach(() => peer.emit('connect'))
+
+      it('dispatches "play" action', () => {
+        expect(play.mock.calls.length).toBe(1)
+      })
+    })
+
+    describe('data', () => {
+
+      beforeEach(() => {
+        window.TextDecoder = class TextDecoder {
+          constructor (encoding) {
+            this.encoding = encoding
+          }
+          decode (object) {
+            return object.toString(this.encoding)
+          }
+        }
+      })
+
+      it('decodes a message', () => {
+        const message = 'test'
+        const object = JSON.stringify({ message })
+        peer.emit('data', Buffer.from(object, 'utf-8'))
+        const { notifications } = store.getState()
+        const keys = Object.keys(notifications)
+        const n = notifications[keys[keys.length - 1]]
+        expect(n).toEqual({
+          id: jasmine.any(String),
+          type: 'info',
+          message: `${user.id}: ${message}`
+        })
+      })
+    })
+  })
+
+  describe('get', () => {
+    it('returns undefined when not found', () => {
+      const { peers } = store.getState()
+      expect(peers[user.id]).not.toBeDefined()
+    })
+
+    it('returns Peer instance when found', () => {
+      store.dispatch(
+        PeerActions.createPeer({ socket, user, initiator: 'user2', stream })
+      )
+
+      const { peers } = store.getState()
+      expect(peers[user.id]).toBe(Peer.instances[0])
+    })
+  })
+
+  describe('destroyPeers', () => {
+    it('destroys all peers and removes them', () => {
+      store.dispatch(PeerActions.createPeer({
+        socket, user: { id: 'user2' }, initiator: 'user2', stream
+      }))
+      store.dispatch(PeerActions.createPeer({
+        socket, user: { id: 'user3' }, initiator: 'user3', stream
+      }))
+
+      store.dispatch(PeerActions.destroyPeers())
+
+      expect(Peer.instances[0].destroy.mock.calls.length).toEqual(1)
+      expect(Peer.instances[1].destroy.mock.calls.length).toEqual(1)
+
+      const { peers } = store.getState()
+      expect(Object.keys(peers)).toEqual([])
+    })
+  })
+
+  describe('sendMessage', () => {
+
+    beforeEach(() => {
+      store.dispatch(PeerActions.createPeer({
+        socket, user: { id: 'user2' }, initiator: 'user2', stream
+      }))
+      store.dispatch(PeerActions.createPeer({
+        socket, user: { id: 'user3' }, initiator: 'user3', stream
+      }))
+    })
+
+    it('sends a message to all peers', () => {
+      store.dispatch(PeerActions.sendMessage('test'))
+      const { peers } = store.getState()
+      expect(peers['user2'].send.mock.calls)
+      .toEqual([[ '{"message":"test"}' ]])
+      expect(peers['user3'].send.mock.calls)
+      .toEqual([[ '{"message":"test"}' ]])
+    })
+
+  })
+})

src/client/actions/__tests__/SocketActions-test.js

@@ -1,46 +1,46 @@
 jest.mock('simple-peer')
-jest.mock('../../store.js')
 jest.mock('../../callId.js')
 jest.mock('../../iceServers.js')
+jest.mock('../../window/createObjectURL.js')
 
+import * as SocketActions from '../SocketActions.js'
 import * as constants from '../../constants.js'
-import handshake from '../handshake.js'
 import Peer from 'simple-peer'
-import peers from '../peers.js'
+import reducers from '../../reducers/index.js'
-import store from '../../store.js'
 import { EventEmitter } from 'events'
+import { createStore } from '../../store.js'
 
-describe('handshake', () => {
+describe('SocketActions', () => {
-  let socket
+  const roomName = 'bla'
+
+  let socket, store
   beforeEach(() => {
     socket = new EventEmitter()
     socket.id = 'a'
 
+    store = createStore()
+
     Peer.instances = []
-    store.clearActions()
   })
 
-  afterEach(() => peers.clear())
+  describe('handshake', () => {
-
-  describe('socket events', () => {
     describe('users', () => {
-      it('add a peer for each new user and destroy peers for missing', () => {
+      beforeEach(() => {
-        handshake({ socket, roomName: 'bla' })
+        store.dispatch(SocketActions.handshake({ socket, roomName }))
-
+        const payload = {
-        // given
-        let payload = {
           users: [{ id: 'a' }, { id: 'b' }],
           initiator: 'a'
         }
         socket.emit('users', payload)
         expect(Peer.instances.length).toBe(1)
+      })
 
-        // when
+      it('adds a peer for each new user and destroys peers for missing', () => {
-        payload = {
+        const payload = {
           users: [{ id: 'a' }, { id: 'c' }],
           initiator: 'c'
         }
-        socket.emit('users', payload)
+        socket.emit(constants.SOCKET_EVENT_USERS, payload)
 
         // then
         expect(Peer.instances.length).toBe(2)
@@ -53,7 +53,7 @@ describe('handshake', () => {
       let data
       beforeEach(() => {
         data = {}
-        handshake({ socket, roomName: 'bla' })
+        store.dispatch(SocketActions.handshake({ socket, roomName }))
         socket.emit('users', {
           initiator: 'a',
           users: [{ id: 'a' }, { id: 'b' }]
@@ -88,7 +88,7 @@ describe('handshake', () => {
       let ready = false
       socket.once('ready', () => { ready = true })
 
-      handshake({ socket, roomName: 'bla' })
+      store.dispatch(SocketActions.handshake({ socket, roomName }))
 
       socket.emit('users', {
         initiator: 'a',
@@ -102,7 +102,7 @@ describe('handshake', () => {
 
     describe('error', () => {
       it('destroys peer', () => {
-        peer.emit('error', new Error('bla'))
+        peer.emit(constants.PEER_EVENT_ERROR, new Error('bla'))
         expect(peer.destroy.mock.calls.length).toBe(1)
       })
     })
@@ -123,38 +123,29 @@ describe('handshake', () => {
 
     describe('stream', () => {
       it('adds a stream to streamStore', () => {
-        store.clearActions()
+        const stream = {}
-        let stream = {}
+        peer.emit(constants.PEER_EVENT_STREAM, stream)
-        peer.emit('stream', stream)
 
-        expect(store.getActions()).toEqual([{
+        expect(store.getState().streams).toEqual({
-          type: constants.STREAM_ADD,
+          b: jasmine.any(String)
-          payload: {
+        })
-            stream,
-            userId: 'b'
-          }
-        }])
       })
     })
 
     describe('close', () => {
-      it('removes stream from streamStore', () => {
+      beforeEach(() => {
-        store.clearActions()
+        const stream = {}
-        peer.emit('close')
+        peer.emit(constants.PEER_EVENT_STREAM, stream)
+        expect(store.getState().streams).toEqual({
+          b: jasmine.any(String)
+        })
+      })
 
-        expect(store.getActions()).toEqual([{
+      it('removes stream & peer from store', () => {
-          type: constants.NOTIFY,
+        expect(store.getState().peers).toEqual({ b: peer })
-          payload: {
+        peer.emit('close')
-            id: jasmine.any(String),
+        expect(store.getState().streams).toEqual({})
-            message: 'Peer connection closed',
+        expect(store.getState().peers).toEqual({})
-            type: 'error'
-          }
-        }, {
-          type: constants.STREAM_REMOVE,
-          payload: {
-            userId: 'b'
-          }
-        }])
       })
     })
   })

src/client/components/App.js

@@ -3,15 +3,15 @@ import Input from './Input.js'
 import Notifications, { NotificationPropTypes } from './Notifications.js'
 import PropTypes from 'prop-types'
 import React from 'react'
-import Video, { StreamPropType } from './Video.js'
+import Video from './Video.js'
 import _ from 'underscore'
 
 export default class App extends React.Component {
   static propTypes = {
     dismissAlert: PropTypes.func.isRequired,
-    streams: PropTypes.objectOf(StreamPropType).isRequired,
+    streams: PropTypes.objectOf(PropTypes.string).isRequired,
     alerts: PropTypes.arrayOf(AlertPropType).isRequired,
-    activate: PropTypes.func.isRequired,
+    setActive: PropTypes.func.isRequired,
     active: PropTypes.string,
     init: PropTypes.func.isRequired,
     notify: PropTypes.func.isRequired,
@@ -23,19 +23,27 @@ export default class App extends React.Component {
   }
   render () {
     const {
-      active, activate, alerts, dismissAlert, notify, notifications, streams
+      active,
+      alerts,
+      dismissAlert,
+      notifications,
+      notify,
+      sendMessage,
+      setActive,
+      streams
     } = this.props
 
     return (<div className="app">
       <Alerts alerts={alerts} dismiss={dismissAlert} />
       <Notifications notifications={notifications} />
-      <Input notify={notify} />
+      <Input notify={notify} sendMessage={sendMessage} />
       <div className="videos">
         {_.map(streams, (stream, userId) => (
           <Video
-            activate={activate}
+            setActive={setActive}
             active={userId === active}
             key={userId}
+            userId={userId}
             stream={stream}
           />
         ))}

src/client/components/Input.js

@@ -1,10 +1,10 @@
 import PropTypes from 'prop-types'
 import React from 'react'
-import peers from '../peer/peers.js'
 
 export default class Input extends React.Component {
   static propTypes = {
-    notify: PropTypes.func.isRequired
+    notify: PropTypes.func.isRequired,
+    sendMessage: PropTypes.func.isRequired,
   }
   constructor () {
     super()
@@ -28,10 +28,10 @@ export default class Input extends React.Component {
     }
   }
   submit = () => {
-    const { notify } = this.props
+    const { notify, sendMessage } = this.props
     const { message } = this.state
-    peers.message(message)
     notify('You: ' + message)
+    sendMessage(message)
     this.setState({ message: '' })
   }
   render () {

src/client/components/Video.js

@@ -3,36 +3,32 @@ import React from 'react'
 import classnames from 'classnames'
 import { ME } from '../constants.js'
 
-export const StreamPropType = PropTypes.shape({
-  userId: PropTypes.string.isRequired,
-  url: PropTypes.string.isRequired
-})
-
 export default class Video extends React.Component {
   static propTypes = {
-    activate: PropTypes.func.isRequired,
+    setActive: PropTypes.func.isRequired,
     active: PropTypes.bool.isRequired,
-    stream: StreamPropType.isRequired
+    stream: PropTypes.string.isRequired,
+    userId: PropTypes.string.isRequired
   }
-  activate = e => {
+  setActive = e => {
-    const { activate, stream: { userId } } = this.props
+    const { setActive, userId } = this.props
     this.play(e)
-    activate(userId)
+    setActive(userId)
   }
   play = e => {
     e.preventDefault()
     e.target.play()
   }
   render () {
-    const { active, stream: { userId, url } } = this.props
+    const { active, stream, userId } = this.props
     const className = classnames('video-container', { active })
     return (
       <div className={className}>
         <video
           muted={userId === ME}
-          onClick={this.activate}
+          onClick={this.setActive}
           onLoadedMetadata={this.play}
-          src={url}
+          src={stream}
         />
       </div>
     )

src/client/components/__tests__/Input-test.js

@@ -1,20 +1,20 @@
 jest.mock('../../callId.js')
 jest.mock('../../iceServers.js')
-jest.mock('../../peer/peers.js')
 
 import Input from '../Input.js'
 import React from 'react'
 import ReactDOM from 'react-dom'
 import TestUtils from 'react-dom/test-utils'
-import peers from '../../peer/peers.js'
 
 describe('components/Input', () => {
 
-  let component, node, notify
+  let component, node, notify, sendMessage
   function render () {
     notify = jest.fn()
+    sendMessage = jest.fn()
     component = TestUtils.renderIntoDocument(
       <Input
+        sendMessage={sendMessage}
         notify={notify}
       />
     )
@@ -28,7 +28,7 @@ describe('components/Input', () => {
 
     let input
     beforeEach(() => {
-      peers.message.mockClear()
+      sendMessage.mockClear()
       input = node.querySelector('input')
       TestUtils.Simulate.change(input, {
         target: { value: message }
@@ -40,7 +40,7 @@ describe('components/Input', () => {
       it('sends a message', () => {
         TestUtils.Simulate.submit(node)
         expect(input.value).toBe('')
-        expect(peers.message.mock.calls).toEqual([[ message ]])
+        expect(sendMessage.mock.calls).toEqual([[ message ]])
         expect(notify.mock.calls).toEqual([[ `You: ${message}` ]])
       })
     })
@@ -51,7 +51,7 @@ describe('components/Input', () => {
           key: 'Enter'
         })
         expect(input.value).toBe('')
-        expect(peers.message.mock.calls).toEqual([[ message ]])
+        expect(sendMessage.mock.calls).toEqual([[ message ]])
         expect(notify.mock.calls).toEqual([[ `You: ${message}` ]])
       })
 
@@ -59,7 +59,7 @@ describe('components/Input', () => {
         TestUtils.Simulate.keyPress(input, {
           key: 'test'
         })
-        expect(peers.message.mock.calls.length).toBe(0)
+        expect(sendMessage.mock.calls.length).toBe(0)
         expect(notify.mock.calls.length).toBe(0)
       })
     })

src/client/constants.js

@@ -1,19 +1,34 @@
-import { PENDING, FULFILLED, REJECTED } from 'redux-promise-middleware'
-export const ME = '_me_'
 
-export const INIT = 'INIT'
+export const ACTIVE_SET = 'ACTIVE_SET'
-export const INIT_PENDING = `${INIT}_${PENDING}`
-export const INIT_FULFILLED = `${INIT}_${FULFILLED}`
-export const INIT_REJECTED = `${INIT}_${REJECTED}`
 
 export const ALERT = 'ALERT'
 export const ALERT_DISMISS = 'ALERT_DISMISS'
 export const ALERT_CLEAR = 'ALERT_CLEAR'
 
+export const INIT = 'INIT'
+export const INIT_PENDING = `${INIT}_PENDING`
+export const INIT_FULFILLED = `${INIT}_FULFILLED`
+export const INIT_REJECTED = `${INIT}_REJECTED`
+
+export const ME = '_me_'
+
 export const NOTIFY = 'NOTIFY'
 export const NOTIFY_DISMISS = 'NOTIFY_DISMISS'
 export const NOTIFY_CLEAR = 'NOTIFY_CLEAR'
 
-export const STREAM_ADD = 'STREAM_ADD'
+export const PEER_ADD = 'PEER_ADD'
-export const STREAM_ACTIVATE = 'STREAM_ACTIVATE'
+export const PEER_REMOVE = 'PEER_REMOVE'
-export const STREAM_REMOVE = 'STREAM_REMOVE'
+export const PEERS_DESTROY = 'PEERS_DESTROY'
+
+export const PEER_EVENT_ERROR = 'error'
+export const PEER_EVENT_CONNECT = 'connect'
+export const PEER_EVENT_CLOSE = 'close'
+export const PEER_EVENT_SIGNAL = 'signal'
+export const PEER_EVENT_STREAM = 'stream'
+export const PEER_EVENT_DATA = 'data'
+
+export const SOCKET_EVENT_SIGNAL = 'signal'
+export const SOCKET_EVENT_USERS = 'users'
+
+export const STREAM_ADD = 'PEER_STREAM_ADD'
+export const STREAM_REMOVE = 'PEER_STREAM_REMOVE'

src/client/containers/App.js

@@ -1,5 +1,6 @@
 import * as CallActions from '../actions/CallActions.js'
 import * as NotifyActions from '../actions/NotifyActions.js'
+import * as PeerActions from '../actions/PeerActions.js'
 import * as StreamActions from '../actions/StreamActions.js'
 import App from '../components/App.js'
 import { bindActionCreators } from 'redux'
@@ -7,7 +8,7 @@ import { connect } from 'react-redux'
 
 function mapStateToProps (state) {
   return {
-    streams: state.streams.all,
+    streams: state.streams,
     alerts: state.alerts,
     notifications: state.notifications,
     active: state.streams.active
@@ -16,7 +17,8 @@ function mapStateToProps (state) {
 
 function mapDispatchToProps (dispatch) {
   return {
-    activate: bindActionCreators(StreamActions.activateStream, dispatch),
+    setActive: bindActionCreators(StreamActions.setActive, dispatch),
+    sendMessage: bindActionCreators(PeerActions.sendMessage, dispatch),
     dismissAlert: bindActionCreators(NotifyActions.dismissAlert, dispatch),
     init: bindActionCreators(CallActions.init, dispatch),
     notify: bindActionCreators(NotifyActions.info, dispatch)

src/client/peer/__tests__/peers-test.js

@@ -1,228 +0,0 @@
-jest.mock('../../window/video.js')
-jest.mock('../../callId.js')
-jest.mock('../../iceServers.js')
-jest.mock('../../store.js')
-  // const configureStore = require('redux-mock-store').default
-  // const { middlewares } = require('../../middlewares.js')
-  // return configureStore(middlewares)({})
-// })
-jest.mock('simple-peer')
-  // const EventEmitter = require('events').EventEmitter
-  // const Peer = jest.genMockFunction().mockImplementation(() => {
-  //   let peer = new EventEmitter()
-  //   peer.destroy = jest.genMockFunction()
-  //   peer.signal = jest.genMockFunction()
-  //   Peer.instances.push(peer)
-  //   return peer
-  // })
-  // Peer.instances = []
-  // return Peer
-// })
-
-import * as constants from '../../constants.js'
-import Peer from 'simple-peer'
-import peers from '../peers.js'
-import store from '../../store.js'
-import { EventEmitter } from 'events'
-import { play } from '../../window/video.js'
-
-describe('peers', () => {
-  function createSocket () {
-    const socket = new EventEmitter()
-    socket.id = 'user1'
-    return socket
-  }
-
-  let socket, stream, user
-  beforeEach(() => {
-    store.clearActions()
-
-    user = { id: 'user2' }
-    socket = createSocket()
-    Peer.instances = []
-    Peer.mockClear()
-    play.mockClear()
-    stream = { stream: true }
-  })
-
-  const actions = {
-    connecting: {
-      type: constants.NOTIFY,
-      payload: {
-        id: jasmine.any(String),
-        message: 'Connecting to peer...',
-        type: 'warning'
-      }
-    },
-    established: {
-      type: constants.NOTIFY,
-      payload: {
-        id: jasmine.any(String),
-        message: 'Peer connection established',
-        type: 'warning'
-      }
-    }
-  }
-
-  afterEach(() => peers.clear())
-
-  describe('create', () => {
-    it('creates a new peer', () => {
-      peers.create({ socket, user, initiator: 'user2', stream })
-
-      expect(store.getActions()).toEqual([actions.connecting])
-
-      expect(Peer.instances.length).toBe(1)
-      expect(Peer.mock.calls.length).toBe(1)
-      expect(Peer.mock.calls[0][0].initiator).toBe(false)
-      expect(Peer.mock.calls[0][0].stream).toBe(stream)
-    })
-
-    it('sets initiator correctly', () => {
-      peers.create({ socket, user, initiator: 'user1', stream })
-
-      expect(Peer.instances.length).toBe(1)
-      expect(Peer.mock.calls.length).toBe(1)
-      expect(Peer.mock.calls[0][0].initiator).toBe(true)
-      expect(Peer.mock.calls[0][0].stream).toBe(stream)
-    })
-
-    it('destroys old peer before creating new one', () => {
-      peers.create({ socket, user, initiator: 'user2', stream })
-      peers.create({ socket, user, initiator: 'user2', stream })
-
-      expect(Peer.instances.length).toBe(2)
-      expect(Peer.mock.calls.length).toBe(2)
-      expect(Peer.instances[0].destroy.mock.calls.length).toBe(1)
-      expect(Peer.instances[1].destroy.mock.calls.length).toBe(0)
-    })
-  })
-
-  describe('events', () => {
-    let peer
-
-    beforeEach(() => {
-      peers.create({ socket, user, initiator: 'user1', stream })
-      peer = Peer.instances[0]
-    })
-
-    describe('connect', () => {
-      beforeEach(() => peer.emit('connect'))
-
-      it('sends a notification', () => {
-        expect(store.getActions()).toEqual([
-          actions.connecting,
-          actions.established
-        ])
-      })
-
-      it('dispatches "play" action', () => {
-        expect(play.mock.calls.length).toBe(1)
-      })
-    })
-
-    describe('data', () => {
-
-      beforeEach(() => {
-        window.TextDecoder = class TextDecoder {
-          constructor (encoding) {
-            this.encoding = encoding
-          }
-          decode (object) {
-            return object.toString(this.encoding)
-          }
-        }
-      })
-
-      it('decodes a message', () => {
-        store.clearActions()
-        const message = 'test'
-        const object = JSON.stringify({ message })
-        peer.emit('data', Buffer.from(object, 'utf-8'))
-        expect(store.getActions()).toEqual([{
-          type: constants.NOTIFY,
-          payload: {
-            id: jasmine.any(String),
-            type: 'info',
-            message: `${user.id}: ${message}`
-          }
-        }])
-      })
-    })
-  })
-
-  describe('get', () => {
-    it('returns undefined when not found', () => {
-      expect(peers.get(user.id)).not.toBeDefined()
-    })
-
-    it('returns Peer instance when found', () => {
-      peers.create({ socket, user, initiator: 'user2', stream })
-
-      expect(peers.get(user.id)).toBe(Peer.instances[0])
-    })
-  })
-
-  describe('getIds', () => {
-    it('returns ids of all peers', () => {
-      peers.create({
-        socket, user: { id: 'user2' }, initiator: 'user2', stream
-      })
-      peers.create({
-        socket, user: { id: 'user3' }, initiator: 'user3', stream
-      })
-
-      expect(peers.getIds()).toEqual([ 'user2', 'user3' ])
-    })
-  })
-
-  describe('destroy', () => {
-    it('destroys a peer and removes it', () => {
-      peers.create({ socket, user, initiator: 'user2', stream })
-
-      peers.destroy(user.id)
-
-      expect(Peer.instances[0].destroy.mock.calls.length).toEqual(1)
-    })
-
-    it('throws no error when peer missing', () => {
-      peers.destroy('bla123')
-    })
-  })
-
-  describe('clear', () => {
-    it('destroys all peers and removes them', () => {
-      peers.create({
-        socket, user: { id: 'user2' }, initiator: 'user2', stream
-      })
-      peers.create({
-        socket, user: { id: 'user3' }, initiator: 'user3', stream
-      })
-
-      peers.clear()
-
-      expect(Peer.instances[0].destroy.mock.calls.length).toEqual(1)
-      expect(Peer.instances[1].destroy.mock.calls.length).toEqual(1)
-
-      expect(peers.getIds()).toEqual([])
-    })
-  })
-
-  describe('message', () => {
-
-    it('sends a message to all peers', () => {
-      peers.create({
-        socket, user: { id: 'user2' }, initiator: 'user2', stream
-      })
-      peers.create({
-        socket, user: { id: 'user3' }, initiator: 'user3', stream
-      })
-      peers.message('test')
-      expect(peers.get('user2').send.mock.calls)
-      .toEqual([[ '{"message":"test"}' ]])
-      expect(peers.get('user3').send.mock.calls)
-      .toEqual([[ '{"message":"test"}' ]])
-    })
-
-  })
-})

src/client/peer/handshake.js

@@ -1,47 +0,0 @@
-import * as NotifyActions from '../actions/NotifyActions.js'
-import _ from 'underscore'
-import _debug from 'debug'
-import peers from './peers.js'
-import store from '../store.js'
-
-const debug = _debug('peercalls')
-const { dispatch } = store
-
-export default function handshake ({ socket, roomName, stream }) {
-  function createPeer (user, initiator) {
-    return peers.create({ socket, user, initiator, stream })
-  }
-
-  socket.on('signal', payload => {
-    let peer = peers.get(payload.userId)
-    let signal = payload.signal
-    // debug('socket signal, userId: %s, signal: %o', payload.userId, signal);
-
-    if (!peer) return debug('user: %s, no peer found', payload.userId)
-    peer.signal(signal)
-  })
-
-  socket.on('users', payload => {
-    let { initiator, users } = payload
-    debug('socket users: %o', users)
-    dispatch(
-      NotifyActions.info('Connected users: {0}', users.length)
-    )
-
-    users
-    .filter(user => !peers.get(user.id) && user.id !== socket.id)
-    .forEach(user => createPeer(user, initiator))
-
-    let newUsersMap = _.indexBy(users, 'id')
-    peers.getIds()
-    .filter(id => !newUsersMap[id])
-    .forEach(peers.destroy)
-  })
-
-  debug('socket.id: %s', socket.id)
-  debug('emit ready for room: %s', roomName)
-  dispatch(
-    NotifyActions.info('Ready for connections')
-  )
-  socket.emit('ready', roomName)
-}

src/client/peer/peers.js

@@ -1,120 +0,0 @@
-import * as NotifyActions from '../actions/NotifyActions.js'
-import * as StreamActions from '../actions/StreamActions.js'
-import Peer from 'simple-peer'
-import _ from 'underscore'
-import _debug from 'debug'
-import iceServers from '../iceServers.js'
-import store from '../store.js'
-import { play } from '../window/video.js'
-
-const debug = _debug('peercalls')
-const { dispatch } = store
-
-let peers = {}
-
-/**
- * @param {Socket} socket
- * @param {User} user
- * @param {String} user.id
- * @param {Boolean} [initiator=false]
- * @param {MediaStream} [stream]
- */
-function create ({ socket, user, initiator, stream }) {
-  debug('create peer: %s, stream:', user.id, stream)
-  dispatch(
-    NotifyActions.warning('Connecting to peer...')
-  )
-
-  if (peers[user.id]) {
-    dispatch(
-      NotifyActions.info('Cleaning up old connection...')
-    )
-    destroy(user.id)
-  }
-
-  const peer = peers[user.id] = new Peer({
-    initiator: socket.id === initiator,
-    stream,
-    config: { iceServers }
-  })
-
-  peer.once('error', err => {
-    debug('peer: %s, error %s', user.id, err.stack)
-    dispatch(
-      NotifyActions.error('A peer connection error occurred')
-    )
-    destroy(user.id)
-  })
-
-  peer.on('signal', signal => {
-    debug('peer: %s, signal: %o', user.id, signal)
-
-    const payload = { userId: user.id, signal }
-    socket.emit('signal', payload)
-  })
-
-  peer.once('connect', () => {
-    debug('peer: %s, connect', user.id)
-    dispatch(
-      NotifyActions.warning('Peer connection established')
-    )
-    play()
-  })
-
-  peer.on('stream', stream => {
-    debug('peer: %s, stream', user.id)
-    dispatch(StreamActions.addStream({
-      userId: user.id,
-      stream
-    }))
-  })
-
-  peer.on('data', object => {
-    object = JSON.parse(new window.TextDecoder('utf-8').decode(object))
-    debug('peer: %s, message: %o', user.id, object)
-    dispatch(
-      NotifyActions.info('' + user.id + ': ' + object.message)
-    )
-  })
-
-  peer.once('close', () => {
-    debug('peer: %s, close', user.id)
-    dispatch(
-      NotifyActions.error('Peer connection closed')
-    )
-    dispatch(
-      StreamActions.removeStream(user.id)
-    )
-
-    delete peers[user.id]
-  })
-}
-
-function get (userId) {
-  return peers[userId]
-}
-
-function getIds () {
-  return _.map(peers, (peer, id) => id)
-}
-
-function clear () {
-  debug('clear')
-  _.each(peers, (_, userId) => destroy(userId))
-  peers = {}
-}
-
-function destroy (userId) {
-  debug('destroy peer: %s', userId)
-  let peer = peers[userId]
-  if (!peer) return debug('peer: %s peer not found', userId)
-  peer.destroy()
-  delete peers[userId]
-}
-
-function message (message) {
-  message = JSON.stringify({ message })
-  _.each(peers, peer => peer.send(message))
-}
-
-module.exports = { create, get, getIds, destroy, clear, message }

src/client/reducers/__tests__/active-test.js

@@ -0,0 +1,16 @@
+import * as constants from '../../constants.js'
+import active from '../active.js'
+
+describe('reducers/active', () => {
+
+  it('sets active to userId', () => {
+    const userId = 'test'
+    let state = active()
+    state = active(state, {
+      type: constants.ACTIVE_SET,
+      payload: { userId }
+    })
+    expect(state).toBe(userId)
+  })
+
+})

src/client/reducers/__tests__/streams-test.js

@@ -22,10 +22,7 @@ describe('reducers/alerts', () => {
 
   describe('defaultState', () => {
     it('should have default state set', () => {
-      expect(store.getState().streams).toEqual({
+      expect(store.getState().streams).toEqual({})
-        active: null,
-        all: {}
-      })
     })
   })
 
@@ -33,13 +30,7 @@ describe('reducers/alerts', () => {
     it('adds a stream', () => {
       store.dispatch(StreamActions.addStream({ userId, stream }))
       expect(store.getState().streams).toEqual({
-        active: userId,
+        [userId]: jasmine.any(String)
-        all: {
-          [userId]: {
-            userId,
-            url: jasmine.any(String)
-          }
-        }
       })
     })
   })
@@ -48,20 +39,7 @@ describe('reducers/alerts', () => {
     it('removes a stream', () => {
       store.dispatch(StreamActions.addStream({ userId, stream }))
       store.dispatch(StreamActions.removeStream(userId))
-      expect(store.getState().streams).toEqual({
+      expect(store.getState().streams).toEqual({})
-        active: userId,
-        all: {}
-      })
-    })
-  })
-
-  describe('activateStream', () => {
-    it('activates a stream', () => {
-      store.dispatch(StreamActions.activateStream(userId))
-      expect(store.getState().streams).toEqual({
-        active: userId,
-        all: {}
-      })
     })
   })
 

src/client/reducers/active.js

@@ -0,0 +1,10 @@
+import * as constants from '../constants.js'
+
+export default function active (state = null, action) {
+  switch (action && action.type) {
+    case constants.ACTIVE_SET:
+      return action.payload.userId
+    default:
+      return state
+  }
+}

src/client/reducers/index.js

@@ -1,10 +1,14 @@
+import active from './active.js'
 import alerts from './alerts.js'
 import notifications from './notifications.js'
+import peers from './peers.js'
 import streams from './streams.js'
 import { combineReducers } from 'redux'
 
 export default combineReducers({
+  active,
   alerts,
   notifications,
+  peers,
   streams
 })

src/client/reducers/peers.js

@@ -0,0 +1,21 @@
+import * as constants from '../constants.js'
+import _ from 'underscore'
+
+const defaultState = {}
+
+export default function peers (state = defaultState, action) {
+  switch (action && action.type) {
+    case constants.PEER_ADD:
+      return {
+        ...state,
+        [action.payload.userId]: action.payload.peer
+      }
+    case constants.PEER_REMOVE:
+      return _.omit(state, [action.payload.userId])
+    case constants.PEERS_DESTROY:
+      _.each(state, peer => peer.destroy())
+      return defaultState
+    default:
+      return state
+  }
+}

src/client/reducers/streams.js

@@ -2,33 +2,21 @@ import * as constants from '../constants.js'
 import createObjectURL from '../window/createObjectURL'
 import Immutable from 'seamless-immutable'
 
-const defaultState = Immutable({
+const defaultState = Immutable({})
-  active: null,
-  all: {}
-})
 
 function addStream (state, action) {
   const { userId, stream } = action.payload
-  const all = state.all.merge({
+  return state.merge({
-    [userId]: {
+    [userId]: createObjectURL(stream)
-      userId,
-      url: createObjectURL(stream)
-    }
   })
-  return state.merge({ active: userId, all })
 }
 
-function removeStream (state, action) {
+const removeStream = (state, action) => state.without(action.payload.userId)
-  const all = state.all.without(action.payload.userId)
-  return state.merge({ all })
-}
 
 export default function streams (state = defaultState, action) {
   switch (action && action.type) {
     case constants.STREAM_ADD:
       return addStream(state, action)
-    case constants.STREAM_ACTIVATE:
-      return state.merge({ active: action.payload.userId })
     case constants.STREAM_REMOVE:
       return removeStream(state, action)
     default:

src/client/store.js

@@ -1,11 +1,13 @@
 import { create } from './middlewares.js'
 import reducers from './reducers'
-import { applyMiddleware, createStore } from 'redux'
+import { applyMiddleware, createStore as _createStore } from 'redux'
 export const middlewares = create(
   window.localStorage && window.localStorage.log
 )
 
-export default createStore(
+export const createStore = () => _createStore(
   reducers,
   applyMiddleware.apply(null, middlewares)
 )
+
+export default createStore()