import { CommandClasses , ZWaveError , ZWaveErrorCodes } from "@zwave-js/core" ; import { SortedList } from "alcalzone-shared/sorted-list" ; import { Action , ActorRef , ActorRefFrom , assign , AssignAction , createMachine , forwardTo , Interpreter , MachineOptions , PureAction , spawn , StateMachine , } from "xstate" ; import { pure , raise , send , stop } from "xstate/lib/actions" ; import { messageIsPing } from "../commandclass/NoOperationCC" ; import { MessagePriority } from "../message/Constants" ; import type { Message } from "../message/Message" ; import { InterviewStage , NodeStatus } from "../node/Types" ; import { CommandQueueEvent , createCommandQueueMachine , } from "./CommandQueueMachine" ; import type { SerialAPICommandDoneData , SerialAPICommandMachineParams , } from "./SerialAPICommandMachine" ; import type { ServiceImplementations } from "./StateMachineShared" ; import type { Transaction } from "./Transaction" ; import { createTransactionMachine , TransactionMachine , } from "./TransactionMachine" ; import type { ZWaveOptions } from "./ZWaveOptions" ; export type SendDataErrorData = ( SerialAPICommandDoneData & { type : "failure" ; } ) | { type : "failure" ; reason : "node timeout" ; result ? : undefined ; } ; export interface ActiveTransaction { transaction : Transaction ; machine : ActorRefFrom < TransactionMachine > ; } export interface SendThreadContext { queue : SortedList < Transaction > ; commandQueue : ActorRef < any , any > ; activeTransactions : Map < string , ActiveTransaction > ; counter : number ; paused : boolean ; } export type SendThreadEvent = { type : "add" ; transaction : Transaction ; } | { type : "trigger" ; } | { type : "unsolicited" ; message : Message ; } | { type : "sortQueue" ; } | { type : "NIF" ; nodeId : number ; } | { type : "reduce" ; reducer : TransactionReducer ; } | { type : "resend" ; } | { type : "ACK" ; } | { type : "CAN" ; } | { type : "NAK" ; } | { type : "message" ; message : Message ; } | ( CommandQueueEvent & ( { type : "command_success" ; } | { type : "command_failure" ; } | { type : "command_error" ; } ) ) | { type : "pause" | "unpause" ; } | { type : "forward" ; from : string ; to : string ; payload : any ; } | { type : "transaction_done" ; id : string ; } ; export type SendThreadMachine = StateMachine < SendThreadContext , any , SendThreadEvent > ; export type SendThreadInterpreter = Interpreter < SendThreadContext , any , SendThreadEvent > ; export type TransactionReducerResult = { type : "drop" ; } | { type : "keep" ; } | { type : "reject" ; message : string ; code : ZWaveErrorCodes ; } | { type : "resolve" ; message ? : Message ; } | { type : "requeue" ; priority ? : MessagePriority ; tag ? : any ; } ; export type TransactionReducer = ( transaction : Transaction , source : "queue" | "active" ) => TransactionReducerResult ; export type SendThreadMachineParams = { timeouts : SerialAPICommandMachineParams [ "timeouts" ] & Pick < ZWaveOptions [ "timeouts" ] , "report" > ; attempts : SerialAPICommandMachineParams [ "attempts" ] ; } ; const finalizeTransaction = pure ( ( ctx , evt ) => [ stop ( evt . id ) , assign ( ( ctx ) => { const transaction = ctx . activeTransactions . get ( evt . id ) ?. transaction ; if ( transaction ?. pauseSendThread ) ctx . paused = true ; ctx . activeTransactions . delete ( evt . id ) ; return ctx ; } ) as any , ] ) ; const forwardToCommandQueue = forwardTo < any , any > ( ( ctx ) => ctx . commandQueue ) ; const sortQueue = assign ( { queue : ( ctx ) => { const queue = ctx . queue ; const items = [ ... queue ] ; queue . clear ( ) ; queue . add ( ... items ) ; return queue ; } } ) ; const guards = { mayStartTransaction : ( ctx , evt , meta ) => { if ( ctx . paused ) return false ; const nextTransaction = ctx . queue . peekStart ( ) ; if ( ! nextTransaction ) return false ; const message = nextTransaction . message ; const targetNode = message . getNodeUnsafe ( ) ; if ( meta . state . matches ( "busy" ) ) { if ( nextTransaction . priority === MessagePriority . Nonce ) { for ( const active of ctx . activeTransactions . values ( ) ) { if ( active . transaction . message . getNodeId ( ) === nextTransaction . message . getNodeId ( ) ) { return true ; } } } return false ; } if ( nextTransaction . priority === MessagePriority . Nonce || nextTransaction . priority === MessagePriority . Supervision ) { return true ; } if ( messageIsPing ( message ) ) return true ; if ( ! targetNode ) return true ; return ( targetNode . status !== NodeStatus . Asleep || ( ! targetNode . supportsCC ( CommandClasses [ "Wake Up" ] ) && targetNode . interviewStage >= InterviewStage . NodeInfo ) ) ; } , hasNoActiveTransactions : ( ctx ) => ctx . activeTransactions . size === 0 } ; export function createSendThreadMachine ( implementations , params ) { const notifyUnsolicited = ( _ , evt ) => { implementations . notifyUnsolicited ( evt . message ) ; } ; const reduce = pure ( ( ctx , evt ) => { const dropQueued = [ ] ; const stopActive = [ ] ; const requeue = [ ] ; const reduceTransaction = ( transaction , source ) => { const reducerResult = ( evt as SendThreadEvent & { type : "reduce" ; } ) . reducer ( transaction , source ) ; switch ( reducerResult . type ) { case "drop" : ( source === "queue" ? dropQueued : stopActive ) . push ( transaction ) ; break ; case "requeue" : if ( reducerResult . priority != undefined ) { transaction . priority = reducerResult . priority ; } if ( reducerResult . tag != undefined ) { transaction . tag = reducerResult . tag ; } if ( source === "active" ) stopActive . push ( transaction ) ; requeue . push ( transaction ) ; break ; case "resolve" : implementations . resolveTransaction ( transaction , reducerResult . message ) ; ( source === "queue" ? dropQueued : stopActive ) . push ( transaction ) ; break ; case "reject" : implementations . rejectTransaction ( transaction , new ZWaveError ( reducerResult . message , reducerResult . code , undefined , transaction . stack ) ) ; ( source === "queue" ? dropQueued : stopActive ) . push ( transaction ) ; break ; } } ; const { queue , activeTransactions } = ctx ; for ( const transaction of queue ) { reduceTransaction ( transaction , "queue" ) ; } for ( const { transaction } of activeTransactions . values ( ) ) { reduceTransaction ( transaction , "active" ) ; } queue . remove ( ... dropQueued , ... requeue ) ; queue . add ( ... requeue . map ( ( t ) => t . clone ( ) ) ) ; return [ assign ( ( ctx ) => ( { ... ctx , queue } ) ) , ... stopActive . map ( ( t ) => send < SendThreadContext , any , any > ( { type : "remove" , transaction : t } , { to : ctx . commandQueue as any } ) ) , ] ; } ) ; const spawnTransaction = assign ( ( ctx ) => { const newCounter = ( ctx . counter + 1 ) % 0xffffffff ; const id = "T" + newCounter . toString ( 16 ) . padStart ( 8 , "0" ) ; const transaction = ctx . queue . shift ( ) ! ; const machine = spawn ( createTransactionMachine ( id , transaction , implementations ) , { name : id } ) ; ctx . activeTransactions . set ( id , { machine , transaction } ) ; return { ... ctx , counter : newCounter } ; } ) ; const ret = createMachine < SendThreadContext , SendThreadEvent > ( { id : "SendThread" , initial : "init" , preserveActionOrder : true , context : { commandQueue : undefined as any , queue : new SortedList ( ) , activeTransactions : new Map ( ) , counter : 0 , paused : false } , on : { ACK : { actions : forwardToCommandQueue } , CAN : { actions : forwardToCommandQueue } , NAK : { actions : forwardToCommandQueue } , message : { actions : forwardToCommandQueue } , NIF : { actions : pure ( ( ctx , evt ) => { const activeTransactionMachinesForNode = [ ... ctx . activeTransactions . values ( ) , ] . filter ( ( { transaction } ) => transaction . message . getNodeId ( ) === evt . nodeId ) . map ( ( a ) => a . machine . id ) ; return [ ... activeTransactionMachinesForNode . map ( ( id ) => send ( evt , { to : id } ) as any ) , sortQueue , raise ( "trigger" ) as any , ] ; } ) } , add : { actions : [ assign ( { queue : ( ctx , evt ) => { ctx . queue . add ( evt . transaction ) ; return ctx . queue ; } } ) , raise ( "trigger" ) as any , ] } , reduce : { actions : [ reduce , raise ( "trigger" ) as any ] } , unsolicited : { actions : notifyUnsolicited } , sortQueue : { actions : [ sortQueue , raise ( "trigger" ) as any ] } , pause : { actions : [ assign ( { paused : ( ) => true } ) as any ] } , unpause : { actions : [ assign ( { paused : ( ) => false } ) , raise ( "trigger" ) as any , ] } , forward : { actions : send ( ( _ , evt ) => ( { ... evt . payload , from : evt . from } ) , { to : ( _ , evt ) => evt . to } ) } , transaction_done : { actions : [ finalizeTransaction , raise ( "trigger" ) as any ] } } , states : { init : { entry : assign < SendThreadContext , any > ( { commandQueue : ( ) => spawn ( createCommandQueueMachine ( implementations , params ) , { name : "QUEUE" } ) } ) , always : "idle" } , idle : { id : "idle" , always : { cond : "mayStartTransaction" , actions : spawnTransaction , target : "busy" } , on : { trigger : { target : "idle" } } } , busy : { id : "busy" , always : [ { cond : "hasNoActiveTransactions" , target : "idle" } , { cond : "mayStartTransaction" , actions : spawnTransaction , target : "busy" } , ] , on : { trigger : { target : "busy" } } } } } , { guards : { ... guards , every : ( ctx , event , { cond } ) => { const keys = ( cond as any ) . guards as string [ ] ; return keys . every ( ( guardKey ) => guards [ guardKey ] ( ctx , event , undefined as any ) ) ; } } } ) ; return ret ; }