import { eventChannel } from "redux-saga" ; import { put , take } from "redux-saga/effects" ; import { OpcodesForPacket , PacketSet } from "../packet" ; import { IncomingRegistry } from "../registry/incoming" ; import { Action , ActionStreamPacket , ActionStreamPayload } from "./packet" ; /**
 * Listen to an  IncomingRegistry and emit any received actions to the store if they match a pattern
 *
 *  param registry param opcode The opcode to use, must be registered to a  ActionStreamPacket in the  PacketSet param actions export const incomingSaga = < TPacketSet extends PacketSet , TOpcode extends keyof OpcodesForPacket < TPacketSet , ActionStreamPacket > > ( registry , opcode , actions ) => function * ( ) { let expectedPacketIndex = 1 ; const channel = eventChannel < ActionStreamPacket > ( emit => { const onReceiveActions = ( packet ) => emit ( packet ) ; registry . on ( opcode , onReceiveActions ) ; return ( ) => registry . off ( opcode , onReceiveActions ) ; } ) ; const actionQueue = [ ] ; while ( true ) { const { index , actions : [ action ] } = yield take ( channel ) ; const validAction = actions . includes ( action . type ) ; if ( ! validAction ) { console . error ( `Unhandled action type: ${ action . type } (for opcode ${ opcode })` ) ; continue ; } if ( index < expectedPacketIndex ) { console . warn ( `Received packet index ${ index } before lastReceivedPacketIndex ${ expectedPacketIndex } (for opcode ${ opcode })` ) ; } else { actionQueue [ index - expectedPacketIndex ] = action ; while ( actionQueue [ 0 ] ) { const actionFromQueue = actionQueue . shift ( ) ! ; expectedPacketIndex ++ ; yield put ( actionFromQueue ) ; } } } } ;