/* @flow */
import type { MiddlewareAPI } from 'redux';
import { webSocket } from 'rxjs/observable/dom/webSocket';
import { WebSocketSubject } from 'rxjs/observable/dom/WebSocketSubject';
import 'rxjs/add/operator/retry';

const reducer = (state: any = {}, action: any) => {
  switch (action.type) {
    case 'WEBSOCKET_CONNECT':
      return state;
    case 'WEBSOCKET_DISCONNECT':
      return state;
    case 'WEBSOCKET_SEND':
      return {
        ...state,
        payload: action.payload
      };
    case 'WEBSOCKET_PAYLOAD':
      console.log('WEBSOCKET_PAYLOAD');
      return {
        ...state,
        payload: action.payload
      };
    case 'WEBSOCKET_ERROR':
      return {
        ...state,
        error: action.payload
      };
    default:
      return state;
  }
};

const middleware = (next) => (url: ?string) => {
  console.log(next);
  let websocket: WebSocketSubject<any>;

  return (api: MiddlewareAPI<any, any>) => (reduxDispatch: any) => (action: any) => {
    switch (action.type) {
      case 'WEBSOCKET_CONNECT':
        console.log('WEBSOCKET_CONNECT');
        if (!websocket)
          websocket = webSocket(url);

        websocket
          .retry()
          .subscribe(
            msg => { next({ type: 'WEBSOCKET_PAYLOAD', payload: msg }); reduxDispatch({ type: 'WEBSOCKET_PAYLOAD', payload: msg }); },
            err => reduxDispatch({ type: 'WEBSOCKET_ERROR', payload: err })
          );

        return reduxDispatch(action);
      case 'WEBSOCKET_SEND':
        console.log('WEBSOCKET_SEND');
        websocket.next(JSON.stringify(action.payload));
        return reduxDispatch(action);
      case 'WEBSOCKET_DISCONNECT':
        websocket.unsubscribe();
        return reduxDispatch(action);
      default:
        return reduxDispatch(action);
    }
  };
};

export { reducer as WebSocketReducer, middleware as WebSocketMiddleware };

@@ -5,16 +5,73 @@
 flow */%0A
+import type %7B rxweb$Task, Redux$Action %7D from './rxweb';%0A
 import t
@@ -863,16 +863,38 @@
  = (next
+: (value: any) =%3E void
 ) =%3E (ur
@@ -914,29 +914,8 @@
 %3E %7B%0A
-  console.log(next);%0A
   le
@@ -949,16 +949,72 @@
 ct%3Cany%3E;
+%0A  %0A  type RespondFunc = (string, any) =%3E (() =%3E mixed);
 %0A%0A  retu
@@ -1156,50 +1156,8 @@
 T':%0A
-        console.log('WEBSOCKET_CONNECT');%0A
     
@@ -1227,153 +1227,756 @@
     
-websocket%0A          .retry()%0A          .subscribe(%0A            msg =%3E %7B next(%7B type: 'WEBSOCKET_PAYLOAD', payload: msg %7D); reduxDispatch(%7B type: 
+const %7B dispatch, getState %7D = api;%0A        const respond: RespondFunc = (actionType: string, msg: any) =%3E %7B%0A          const task: rxweb$Task = %7B%0A            type: actionType,%0A            data: msg,%0A            next,%0A            init: () =%3E dispatch(%7B ...getState(), type: %60$%7BactionType%7D_INIT%60 %7D),%0A            done: (payload: Redux$Action) =%3E dispatch(%7B ...getState(), payload, type: %60$%7BactionType%7D_SUCCESS%60 %7D),%0A            error: (payload: Redux$Action) =%3E dispatch(%7B ...getState(), payload, type: %60$%7BactionType%7D_ERROR%60 %7D),%0A            store: %7B dispatch, getState %7D%0A          %7D;          %0A          next(task);%0A          return reduxDispatch(task);%0A        %7D;%0A%0A        websocket%0A          .retry()%0A          .subscribe(%0A            msg =%3E respond(
 'WEB
@@ -1996,26 +1996,12 @@
 D', 
-payload: msg %7D); %7D
+msg)
 ,%0A  
@@ -2023,28 +2023,14 @@
 %3E re
-duxDispatch(%7B type: 
+spond(
 'WEB
@@ -2047,23 +2047,12 @@
 OR',
- payload:
  err
- %7D
 )%0A  

Add rxweb$Task to websocket middleware