package Ġcom . eg m . st elli o . search . service Ċ Ċ import Ġarrow . core . Invalid Ċ import Ġarrow . core . Valid Ċ import Ġarrow . core . Validated Ċ import Ġarrow . core . invalid Ċ import Ġarrow . core . valid Ċ import Ġcom . eg m . st elli o . search . model . Attribute Instance Ċ import Ġcom . eg m . st elli o . search . model . Attribute Metadata Ċ import Ġcom . eg m . st elli o . search . model . Temporal Entity Attribute Ċ import Ġcom . eg m . st elli o . search . util . value ToDouble OrNull Ċ import Ġcom . eg m . st elli o . search . util . value ToString OrNull Ċ import Ġcom . eg m . st elli o . shared . model .* Ċ import Ġcom . eg m . st elli o . shared . util . Json Ld Utils . add Context To Element Ċ import Ġcom . eg m . st elli o . shared . util . Json Ld Utils . add Context sTo Entity Ċ import Ġcom . eg m . st elli o . shared . util . Json Ld Utils . compact Term Ċ import Ġcom . eg m . st elli o . shared . util . Json Ld Utils . expand Json Ld Key Ċ import Ġcom . eg m . st elli o . shared . util . Json Utils Ċ import Ġcom . eg m . st elli o . shared . util . Json Utils . deserialize As Ċ import Ġcom . eg m . st elli o . shared . util . Json Utils . serialize Object Ċ import Ġcom . eg m . st elli o . shared . util . RECE IVED _ NON _ PARSE ABLE _ ENTITY Ċ import Ġcom . eg m . st elli o . shared . util . extract Attribute Instance From Compact ed Entity Ċ import Ġcom . eg m . st elli o . shared . util . to Uri Ċ import Ġcom . fasterxml . jackson . databind . JsonNode Ċ import Ġcom . fasterxml . jackson . databind . node . Object Node Ċ import Ġcom . fasterxml . jackson . module . kotlin . jackson ObjectMapper Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . kafka . annotation . Kafka Listener Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġreactor . core . publisher . Mono Ċ import Ġjava . net . URI Ċ import Ġjava . time . Z onedDateTime Ċ Ċ @ Component Ċ class ĠEntity EventListener Service ( ĊĠĠĠ Ġprivate Ġval Ġtemporal Entity Attribute Service : ĠTemp oral Entity Attribute Service , ĊĠĠĠ Ġprivate Ġval Ġattribute Instance Service : ĠAttribute Instance Service Ċ ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( java Class ) ĊĊĠĠĠ Ġ@ Kafka Listener ( topic Pattern Ġ= Ġ" c im . entity .* ", ĠgroupId Ġ= Ġ" context _ search ") ĊĠĠĠ Ġfun Ġprocess Message ( content : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . debug (" Processing Ġmessage : Ġ$ content ") ĊĠĠĠĠĠĠĠ Ġwhen Ġ( val Ġentity Event Ġ= Ġdeserialize As < Entity Event >( content )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠEntity Create Event Ġ-> Ġhandle Entity Create Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠEntity Delete Event Ġ-> Ġhandle Entity Delete Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAttribute Append Event Ġ-> Ġhandle Attribute Append Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAttribute Replace Event Ġ-> Ġhandle Attribute Replace Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAttribute Update Event Ġ-> Ġhandle Attribute Update Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAttribute Delete Event Ġ-> Ġhandle Attribute Delete Event ( entity Event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAttribute Delete All Instances Event Ġ-> Ġhandle Attribute Delete All Instances Event ( entity Event ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Entity Create Event ( entity Create Event : ĠEntity Create Event ) Ġ= ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoperation Payload Ġ= Ġadd Context To Element ( entity Create Event . operation Payload , Ġentity Create Event . contexts ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . create Entity Temporal References ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoperation Payload , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentity Create Event . contexts ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Boot str apped Ġentity Ġ( records Ġcreated : Ġ$ it )") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠBadRequest DataException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( RECE IVED _ NON _ PARSE ABLE _ ENTITY , Ġe ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠInvalid RequestException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( RECE IVED _ NON _ PARSE ABLE _ ENTITY , Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Entity Delete Event ( entity Delete Event : ĠEntity Delete Event ) Ġ= ĊĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . delete Temporal Entity References ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġentity Delete Event . entityId ĊĠĠĠĠĠĠĠ Ġ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Deleted Ġentity Ġ${ entity Delete Event . entityId } Ġ( records Ġdeleted : Ġ$ it )") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Delete Event ( attribute Delete Event : ĠAttribute Delete Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpanded AttributeName Ġ= Ġexpand Json Ld Key ( attribute Delete Event . attribute Name , Ġattribute Delete Event . contexts ) !! ĊĠĠĠĠĠĠĠ Ġval Ġcompact ed Json Ld Entity Ġ= Ġadd Context sTo Entity ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠJson Utils . deserialize Object ( attribute Delete Event . updated Entity ), Ġattribute Delete Event . contexts ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . delete Temporal Attribute References ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpanded AttributeName , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete Event . dataset Id ĊĠĠĠĠĠĠĠ Ġ). zip With ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . update Entity Payload ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialize Object ( compact ed Json Ld Entity ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Deleted Ġtemporal Ġattribute Ġ$ expanded AttributeName Ġwith Ġdataset Id Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ attribute Delete Event . dataset Id } Ġfrom Ġentity Ġ${ attribute Delete Event . entityId } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"( records Ġdeleted : Ġ${ it . t 1 })" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Delete All Instances Event ( ĊĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event : ĠAttribute Delete All Instances Event ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpanded AttributeName Ġ= Ġexpand Json Ld Key ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event . attribute Name , Ġattribute Delete All Instances Event . contexts ĊĠĠĠĠĠĠĠ Ġ) !! ĊĠĠĠĠĠĠĠ Ġval Ġcompact ed Json Ld Entity Ġ= Ġadd Context sTo Entity ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠJson Utils . deserialize Object ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event . updated Entity ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event . contexts ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . delete Temporal Attribute All Instances References ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpanded AttributeName ĊĠĠĠĠĠĠĠ Ġ). zip With ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . update Entity Payload ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Delete All Instances Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialize Object ( compact ed Json Ld Entity ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Deleted Ġall Ġtemporal Ġattributes Ġof Ġ$ expanded AttributeName Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" from Ġentity Ġ${ attribute Delete All Instances Event . entityId } Ġ( records Ġdeleted : Ġ${ it . t 1 })" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Append Event ( attribute Append Event : ĠAttribute Append Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpanded AttributeName Ġ= Ġexpand Json Ld Key ( attribute Append Event . attribute Name , Ġattribute Append Event . contexts ) !! ĊĠĠĠĠĠĠĠ Ġval Ġoperation Payload Node Ġ= Ġj ack son ObjectMapper (). read Tree ( attribute Append Event . operation Payload ) ĊĠĠĠĠĠĠĠ Ġval Ġattribute Node Ġ= Ġoperation Payload Node [ attribute Append Event . attribute Name ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġoperation Payload Node [ expanded AttributeName ] ĊĊĠĠĠĠĠĠĠ Ġif Ġ( attribute Node Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Unable Ġto Ġextract Ġvalues Ġfrom Ġ$ attribute Append Event ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġhandle Attribute Append ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Append Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpanded AttributeName , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Node , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Append Event . updated Entity , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Append Event . contexts ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Replace Event ( attribute Replace Event : ĠAttribute Replace Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpanded AttributeName Ġ= Ġexpand Json Ld Key ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . attribute Name , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . contexts ĊĠĠĠĠĠĠĠ Ġ) !! ĊĠĠĠĠĠĠĠ Ġval Ġoperation Payload Node Ġ= Ġj ack son ObjectMapper (). read Tree ( attribute Replace Event . operation Payload ) ĊĠĠĠĠĠĠĠ Ġval Ġattribute Node Ġ= Ġoperation Payload Node [ attribute Replace Event . attribute Name ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġoperation Payload Node [ expanded AttributeName ] ĊĊĠĠĠĠĠĠĠ Ġhandle Attribute Update ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpanded AttributeName , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . dataset Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Node , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . updated Entity , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Replace Event . contexts ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Update Event ( attribute Update Event : ĠAttribute Update Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpanded AttributeName Ġ= Ġexpand Json Ld Key ( attribute Update Event . attribute Name , Ġattribute Update Event . contexts ) !! ĊĠĠĠĠĠĠĠ Ġval Ġattribute Node Ġ= Ġj ack son ObjectMapper (). read Tree ( attribute Update Event . operation Payload ) ĊĊĠĠĠĠĠĠĠ Ġhandle Attribute Update ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Update Event . entityId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpanded AttributeName , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Update Event . dataset Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Node , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Update Event . updated Entity , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Update Event . contexts ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Attribute Update ( ĊĠĠĠĠĠĠĠ Ġentity Id : ĠURI , ĊĠĠĠĠĠĠĠ Ġexpanded AttributeName : ĠString , ĊĠĠĠĠĠĠĠ Ġdataset Id : ĠURI ?, ĊĠĠĠĠĠĠĠ Ġattribute Values Node : ĠJsonNode , ĊĠĠĠĠĠĠĠ Ġupdated Entity : ĠString , ĊĠĠĠĠĠĠĠ Ġcontexts : ĠList < String > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTODO Ġadd Ġmissing Ġchecks : ĊĠĠĠĠĠĠĠ Ġ// Ġ Ġ- Ġexistence Ġof Ġtemporal Ġentity Ġattribute ĊĊĠĠĠĠĠĠĠ Ġ// Ġreturn Ġearly Ġto Ġavoid Ġextra Ġprocessing Ġif Ġthe Ġattribute Ġis Ġnot Ġa Ġtemporal Ġone ĊĠĠĠĠĠĠĠ Ġif Ġ(! attribute Values Node . has (" observed At ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Ign oring Ġappend Ġevent Ġfor Ġ$ attribute Values Node , Ġit Ġhas Ġno Ġobserved At Ġinformation ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcompact ed Json Ld Entity Ġ= Ġadd Context sTo Entity ( Json Utils . deserialize Object ( updated Entity ), Ġcontexts ) ĊĠĠĠĠĠĠĠ Ġval Ġattribute Instance Payload Ġ= Ġextract Attribute Instance From Compact ed Entity ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompact ed Json Ld Entity , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompact Term ( expanded AttributeName , Ġcontexts ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdataset Id ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ// ĠSince ĠATTRIBUT E _ UPDATE Ġevents Ġpayload Ġmay Ġnot Ġcontain Ġthe Ġattribute Ġtype ĊĠĠĠĠĠĠĠ Ġif Ġ(! attribute Values Node . has (" type ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( attribute Values Node Ġas ĠObject Node ). put (" type ", Ġattribute Instance Payload [" type "] Ġas ĠString ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwhen Ġ( val Ġextracted Attribute Metadata Ġ= Ġto Temporal Attribute Metadata ( attribute Values Node )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvalid Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ( extract ed Attribute Metadata . e ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠValid Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattribute Metadata Ġ= Ġextracted Attribute Metadata . a ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . get For Entity And Attribute ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentity Id , Ġexpanded AttributeName , Ġattribute Metadata . dataset Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ). zip When Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattribute Instance Ġ= ĠAttribute Instance ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Ġ= Ġit , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobserved At Ġ= Ġattribute Metadata . observed At , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġattribute Metadata . value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmeasured Value Ġ= Ġattribute Metadata . meas ured Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpayload Ġ= Ġattribute Instance Payload ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Instance Service . create ( attribute Instance ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. zip When Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . t 2 Ġ!= Ġ- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . update Entity Payload ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentity Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialize Object ( compact ed Json Ld Entity ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ). onError Return (- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĠMono . just (- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . t 1 . t 2 Ġ!= Ġ- 1 Ġ&& Ġit . t 2 Ġ!= Ġ- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Created Ġnew Ġattribute Ġinstance Ġof Ġ$ expanded AttributeName Ġfor Ġ$ entityId ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Failed Ġto Ġpersist Ġnew Ġattribute Ġinstance Ġ$ expanded AttributeName Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" for Ġ$ entityId , Ġignoring Ġit Ġ( insert Ġresults : Ġ${ it . t 1 . t 2 } Ġ/ Ġ${ it . t 2 })) " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġhandle Attribute Append ( ĊĠĠĠĠĠĠĠ Ġentity Id : ĠURI , ĊĠĠĠĠĠĠĠ Ġexpanded AttributeName : ĠString , ĊĠĠĠĠĠĠĠ Ġattribute Values Node : ĠJsonNode , ĊĠĠĠĠĠĠĠ Ġupdated Entity : ĠString , ĊĠĠĠĠĠĠĠ Ġcontexts : ĠList < String > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( val Ġextracted Attribute Metadata Ġ= Ġto Temporal Attribute Metadata ( attribute Values Node )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvalid Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ( extract ed Attribute Metadata . e ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠValid Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattribute Metadata Ġ= Ġextracted Attribute Metadata . a ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcompact ed Json Ld Entity Ġ= Ġadd Context sTo Entity ( Json Utils . deserialize Object ( updated Entity ), Ġcontexts ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtemporal Entity Attribute Ġ= ĠTemp oral Entity Attribute ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentity Id Ġ= Ġentity Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtype Ġ= Ġexpand Json Ld Key ( compact ed Json Ld Entity . getType (), Ġcontexts ) !! , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Name Ġ= Ġexpanded AttributeName , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Type Ġ= Ġattribute Metadata . type , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute ValueType Ġ= Ġattribute Metadata . valueType , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdataset Id Ġ= Ġattribute Metadata . dataset Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattribute Instance Ġ= ĠAttribute Instance ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Ġ= Ġtemporal Entity Attribute . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobserved At Ġ= Ġattribute Metadata . observed At , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmeasured Value Ġ= Ġattribute Metadata . meas ured Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġattribute Metadata . value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpayload Ġ= Ġextract Attribute Instance From Compact ed Entity ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompact ed Json Ld Entity , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompact Term ( expanded AttributeName , Ġcontexts ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Metadata . dataset Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . create ( temporal Entity Attribute ). zip When Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattribute Instance Service . create ( attribute Instance ). then ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemporal Entity Attribute Service . update Entity Payload ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentity Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialize Object ( compact ed Json Ld Entity ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Failed Ġto Ġpersist Ġnew Ġtemporal Ġentity Ġattribute Ġfor Ġ$ entityId Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" with Ġattribute Ġinstance Ġ$ expanded AttributeName , Ġignoring Ġit Ġ(${ it . message })" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Created Ġnew Ġtemporal Ġentity Ġattribute Ġfor Ġ$ entityId Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" with Ġattribute Ġinstance Ġ$ expanded AttributeName " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġto Temporal Attribute Metadata ( json Node : ĠJsonNode ): ĠValid ated < String , ĠAttribute Metadata > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġattribute Type As Text Ġ= Ġjson Node [" type "]. as Text () ĊĠĠĠĠĠĠĠ Ġval Ġattribute Type Ġ= Ġkotlin . run Catch ing Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTemp oral Entity Attribute . AttributeType . valueOf ( attribute Type As Text ) ĊĠĠĠĠĠĠĠ Ġ}. get OrNull () Ġ?: Ġreturn Ġ" Unsupported Ġattribute Ġtype : Ġ$ attribute Type As Text ". invalid () ĊĠĠĠĠĠĠĠ Ġval Ġattribute Value Ġ= Ġwhen Ġ( attribute Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTemp oral Entity Attribute . AttributeType . Relationship Ġ-> ĠPair ( json Node [" object "]. as Text (), Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTemp oral Entity Attribute . AttributeType . Property Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġraw AttributeValue Ġ= Ġjson Node [" value "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġraw AttributeValue Ġ== Ġnull Ġ-> Ġreturn Ġ" Unable Ġto Ġget Ġa Ġvalue Ġfrom Ġattribute : Ġ$ json Node ". invalid () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġraw AttributeValue . isNumber Ġ-> ĠPair ( null , Ġvalue ToDouble OrNull ( raw AttributeValue . as Double ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠPair ( value ToString OrNull ( raw AttributeValue . as Text ()), Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġattribute ValueType Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( attribute Value . second Ġ!= Ġnull ) ĠTemp oral Entity Attribute . Attribute ValueType . MEAS URE ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĠTemp oral Entity Attribute . Attribute ValueType . ANY ĊĊĠĠĠĠĠĠĠ Ġreturn ĠAttribute Metadata ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmeasured Value Ġ= Ġattribute Value . second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġattribute Value . first , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Type Ġ= Ġattribute ValueType , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdataset Id Ġ= Ġjson Node [" dataset Id "] ?. as Text ()?. to Uri (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtype Ġ= Ġattribute Type , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġobserved At Ġ= ĠZ onedDateTime . parse ( json Node [" observed At "]. as Text ()) ĊĠĠĠĠĠĠĠ Ġ). valid () ĊĠĠĠ Ġ} Ċ } Ċ