Websockets integration

[almeynman]

Hi. In your post you have mentioned how you would tackle working with ajax calls. However I was wondering how would you work with websockets.

I was trying to reimplement phoenix-trello, which was written in react/redux/phoenix with rxjs and decided to try your approach. However I cannot figure it out by myself. For instance how would you go about rewriting this code

// js/actions/current_board.js
...
connectToChannel: (socket, boardId) => {
    return dispatch => {
      const channel = socket.channel(`boards:${boardId}`);

      dispatch({ type: Constants.CURRENT_BOARD_FETCHING });

      channel.join().receive('ok', (response) => {
        dispatch({
          type: Constants.BOARDS_SET_CURRENT_BOARD,
          board: response.board,
        });
      });

      channel.on('user:joined', (msg) => {
        dispatch({
          type: Constants.CURRENT_BOARD_CONNECTED_USERS,
          users: msg.users,
        });
      });
  ...

[MichalZalecki]

You can try with Rx.DOM.fromWebSocket or just create your own observable and call observer.next on particular message. There is also fromEventPattern operator.

[almeynman]

Hey Michal. I think Rx.DOM.fromWebSocket is not necessary with phoenix as there is already abstraction on websockets done by phoenix. I have done the following:

// userReducer.js
...
  actions.setCurrentUser$
    .flatMap(user =>
      Observable.fromPromise(AsyncStorage.getItem('token').map(token => ({user, token}))))
    .map(({user, token}) => {
      new Observable(observer => {
        const socket = new Socket('http://localhost:4000/socket', {
          params: { token }
          logger: (kind, msg, data) => console.log(`${kind}: ${msg}`, data)
        })

        socket.connect()

        const channel = socket.channel(`user:${user.id}`)
        channel
          .join()
          .receive('ok', response => observer.next({ type: 'join', socket, channel, currentUser: user }))
          .receive('error', reason => observer.next({ type: 'error', error: reason }))
          .receive('timout', () => observer.next({ type: 'error', error: 'The request has timed out please try againg when you will have internet connection' }))

        connectObserverToChannelJoin(channel, observer)

        channel.on('rooms:add', room => observer.next({ type: 'rooms:add', room }))
      })
    }).map(action => state => {
      switch (action.type) {
        case 'join':
          return { ...state, currentUser: action.currentUser, socket: action.socket, error: null }
        case 'error':
          return { ...state, error: action.error }
        case 'rooms:add':
          // TODO
          return { ...state }
        default:
          return state
      }
    })
...

I think it's quite clean. What are your thought?

[MichalZalecki]

Sure, you could even create separare observable for each event so you don't need to check types so data flow is more declarative. You can use Rx.Subject for that.

[almeynman]

You mean separate action for each event?

[almeynman]

Or it should be done inside new Observable(...)?

[MichalZalecki]

You mean separate action for each event?

You have already separate action for each event :) Separate observable.

Instead of calling observer.next({ type: 'rooms:add'..

const roomsAdd$ = new Rx.Subject;
roomsAdd$.next(room);
return {
  roomsAdd$,
};

Types are not bad though. Pick what suits you best.

[almeynman]

Thanks, subjects are great, will use it!