package Ġpd . sensor . reactive . server . repository Ċ Ċ import Ġkotlinx . coroutines . Flow Preview Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . flatMap Concat Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . on Each Ċ import Ġkotlinx . coroutines . reactive . as Flow Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . data . redis . connection . stream . Map Record Ċ import Ġorg . springframework . data . redis . connection . stream . Record Id Ċ import Ġorg . springframework . data . redis . connection . stream . Stream Offset Ċ import Ġorg . springframework . data . redis . connection . stream . Stream Records Ċ import Ġorg . springframework . data . redis . core . Reactive Redis Template Ċ import Ġorg . springframework . data . redis . core . add And Await Ċ import Ġorg . springframework . data . redis . core . trim And Await Ċ import Ġorg . springframework . data . redis . stream . Stream Receiver Ċ import Ġorg . springframework . stereotype . Repository Ċ import Ġpd . sensor . domain . Sensor Data Ċ Ċ const Ġval ĠST REAM _ KEY Ġ= Ġ" sensor - data " Ċ Ċ @ Repository Ċ class ĠSensor Data Redis Repository ( ĊĠĠĠ Ġval Ġreactive Redis Template : ĠReactive Redis Template < String , ĠSensor Data >, ĊĠĠĠ Ġval Ġstream Receiver : ĠStream Receiver < String , ĠMap Record < String , ĠString , ĠString >> Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( this :: class . java ) ĊĊĠĠĠ Ġ@ Flow Preview ĊĠĠĠ Ġfun Ġsave All ( entity Stream : ĠFlow < Sensor Data >): ĠFlow < Record Id > Ġ= ĊĠĠĠĠĠĠĠ Ġentity Stream ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġto Map Record ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Concat Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Saving Ġrecord : Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreactive Redis Template ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. ops For Stream < String , ĠSensor Data >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġto Map Record ( sensor Data : ĠSensor Data ): ĠMap Record < String , ĠString , ĠString > Ġ= ĊĠĠĠĠĠĠĠ ĠStream Records . new Record () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. ` in `( STREAM _ KEY ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. of Map ( sensor Data . toMap ()) ĊĊĠĠĠ Ġfun Ġread All (): ĠFlow < Sensor Data > Ġ= ĊĠĠĠĠĠĠĠ Ġstream Receiver ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. receive ( Stream Offset . from Start ( STREAM _ KEY )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Each Ġ{ Ġlog . debug (" Received Ġstream Ġrecord : Ġ$ it ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . value . from Map () Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFor Ġtests Ġonly . ĠS aves Ġa Ġsingle Ġitem . ĠMain Ġflow Ġworks Ġwith Ġcor outines ĠFlow ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġsave ( sensor Data : ĠSensor Data ): ĠRecord Id Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġto Map Record ( sensor Data ) ĊĠĠĠĠĠĠĠ Ġlog . debug (" Saving Ġrecord : Ġ$ record ") ĊĠĠĠĠĠĠĠ Ġreturn Ġreactive Redis Template ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. ops For Stream < String , ĠSensor Data >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. add And Await ( record ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFor Ġtests Ġonly ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġdelete All () Ġ{ ĊĠĠĠĠĠĠĠ Ġreactive Redis Template ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. ops For Stream < String , ĠSensor Data >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. trim And Await ( STREAM _ KEY , Ġ 0 ) ĊĠĠĠ Ġ} Ċ }