import 'rxjs/add/observable/from' ; import 'rxjs/add/operator/mergeAll' ; import { Observable } from 'rxjs/Observable' ; import { RequestEvent } from 'snapper-consumer' ; import { Database } from '../db' ; import { Net } from '../Net' ; import { eventParser } from './EventParser' ; import { ParsedWSMsg , WSMsgHandler , WSMsgToDBHandler } from '../utils' ; import { TableInfoByMessageType } from './MapToTable' ; import { Proxy } from './Middleware' ; const methodMap = { 'change' : 'upsert' , 'new' : 'upsert' , 'destroy' : 'delete' , 'remove' : 'delete' } ; export const createMsgHandler = ( proxy = new Proxy ( ) ) => ( msg ) : Observable < any > => { proxy . apply ( msg ) ; return Observable . of ( null ) ; } ; /**
 * refresh 事件需要逐个单独处理
 * destroy 事件没有 data
 */ export const createMsgToDBHandler = ( mapToTable ) => ( msg , db ) : Observable < any > => { const tabInfo = mapToTable . getTableInfo ( msg . type ) ; if ( ! tabInfo ) { return Observable . of ( null ) ; } const { method , id , data } = msg ; const { tabName , pkName } = tabInfo ; const dbMethod = db [ methodMap [ method ] ] ; switch ( method ) { case 'new' : return dbMethod . call ( db , tabName , data ) ; case 'change' : return dbMethod . call ( db , tabName , Array . isArray ( data ) ? data : { ... data , [ pkName ] : id } ) ; case 'destroy' : return dbMethod . call ( db , tabName , { where : { [ pkName ] : id } } ) ; case 'remove' : return dbMethod . call ( db , tabName , { where : Array . isArray ( data ) ? { [ pkName ] : { $in : data } } : { [ pkName ] : data } } ) ; default : return Observable . of ( null ) ; } } ; export function socketHandler ( net , event , handleMsgToDb , handleMsg , mapToTable , db ? ) { const parsedMsgs = eventParser ( event ) ; const signals = parsedMsgs . map ( ( msg ) => { const tabInfo = mapToTable . getTableInfo ( msg . type ) ; const proxyTask$ = handleMsg ( msg ) ; if ( ! tabInfo ) { return proxyTask$ ; } let interceptorsTask$ ; if ( ! db ) { net . bufferSocketPush ( msg ) ; interceptorsTask$ = Observable . of ( null ) ; } else { interceptorsTask$ = handleMsgToDb ( msg , db ) ; } return Observable . merge ( interceptorsTask$ , proxyTask$ ) ; } ) ; return Observable . from ( signals ) . mergeAll ( ) ; }