/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 ĠAlex ĠCh erm en in Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġru . ch erm en in . k io . cep . transforms Ċ Ċ import Ġjava . io . Serializable Ċ import Ġorg . apache . beam . sdk . coders . Serializable Coder Ċ import Ġorg . apache . beam . sdk . state . State Spec Ċ import Ġorg . apache . beam . sdk . state . State Specs Ċ import Ġorg . apache . beam . sdk . state . Value State Ċ import Ġorg . apache . beam . sdk . transforms . Do Fn Ċ import Ġorg . apache . beam . sdk . values . KV Ċ import Ġru . ch erm en in . k io . cep . Complex Event Ċ import Ġru . ch erm en in . k io . cep . n fa . N FA Ċ import Ġru . ch erm en in . k io . cep . pattern . Pattern Ċ Ċ /** Ċ Ġ* ĠOperator Ġto Ġmatch Ġkey - value Ġpairs Ġby Ġpattern . Ċ Ġ* Ċ Ġ* Ġ@ param Ġpattern Ġpattern Ġto Ġmatch Ġvalues Ċ Ġ* Ġ@ param ĠK Ġkey Ġtype Ċ Ġ* Ġ@ param ĠV Ġvalue Ġtype Ċ Ġ*/ Ċ class ĠMatch Value Do Fn < K , ĠV Ġ: ĠSerializable >( private Ġval Ġpattern : ĠPattern < V >) Ġ: ĠDo Fn < KV < K , ĠV >, ĠK V < K , ĠComplex Event < V >> >() Ġ{ ĊĊĠĠĠ Ġ@ State Id ( N FA _ STATE _ KEY ) ĊĠĠĠ Ġval Ġn fa State : ĠState Spec < Value State < N FA <* >>> Ġ= ĠState Specs . value ( Serializable Coder . of ( N FA :: class . java )) ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġ@ Process Element ĊĠĠĠ Ġfun Ġprocess Element ( ĊĠĠĠĠĠĠĠ Ġcontext : ĠProcess Context , ĊĠĠĠĠĠĠĠ Ġ@ State Id ( N FA _ STATE _ KEY ) Ġn fa State : ĠValue State < N FA <* >> ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġn fa Ġ= Ġn fa State . read () Ġ?: Ġpattern . compile () ĊĠĠĠĠĠĠĠ Ġn fa State . write ( n fa ) ĊĠĠĠĠĠĠĠ Ġval Ġelement Ġ= Ġcontext . element () ĊĠĠĠĠĠĠĠ Ġ( n fa Ġas ĠN FA < V >). process ( element . value , Ġcontext . timestamp (). millis ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . output ( KV . of ( element . key , ĠComplex Event ( it ))) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ