// ĠCopyright Ġ 2 0 2 1 ĠThe ĠCross - Media ĠMeasurement ĠAuthors Ċ // Ċ // Ġ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 Ġorg . wf an et . measurement . king dom . deploy . g cloud . spanner . queries Ċ Ċ import Ġcom . google . cloud . spanner . Statement Ċ import Ġorg . wf an et . measurement . g cloud . common . to G cloud Timestamp Ċ import Ġorg . wf an et . measurement . g cloud . spanner . append Clause Ċ import Ġorg . wf an et . measurement . g cloud . spanner . bind Ċ import Ġorg . wf an et . measurement . internal . king dom . Measurement Ċ import Ġorg . wf an et . measurement . internal . king dom . Stream Measure ments Request Ċ import Ġorg . wf an et . measurement . king dom . deploy . g cloud . spanner . readers . Measurement Reader Ċ Ċ class ĠStream Measure ments ( ĊĠ Ġview : ĠMeasurement . View , ĊĠ Ġrequest Filter : ĠStream Measure ments Request . Filter , ĊĠ Ġlimit : ĠInt Ġ= Ġ 0 Ċ ) Ġ: ĠSimple Sp anner Query < Measurement Reader . Result >() Ġ{ ĊĠ Ġoverride Ġval Ġreader Ġ= ĊĠĠĠ ĠMeasurement Reader ( view ). fill Statement Builder Ġ{ ĊĠĠĠĠĠ Ġappend Where Clause ( request Filter ) ĊĠĠĠĠĠ Ġwhen Ġ( view ) Ġ{ ĊĠĠĠĠĠĠĠ ĠMeasurement . View . COMP UT ATION Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġappend Clause (" ORDER ĠBY ĠUpdate Time ĠASC , ĠExternal Computation Id ĠASC ") ĊĠĠĠĠĠĠĠ ĠMeasurement . View . DEFAULT Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġappend Clause (" ORDER ĠBY ĠUpdate Time ĠASC , ĠExternal Measurement Id ĠASC ") ĊĠĠĠĠĠĠĠ ĠMeasurement . View . UN RE CO GN IZED Ġ-> Ġerror (" Unrecognized ĠView ") ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġif Ġ( limit Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠ Ġappend Clause (" LIMIT Ġ@$ LIMIT _ PARAM ") ĊĠĠĠĠĠĠĠ Ġbind ( LIMIT _ PARAM Ġto Ġlimit . to Long ()) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġfun ĠStatement . Builder . append Where Clause ( filter : ĠStream Measure ments Request . Filter ) Ġ{ ĊĠĠĠ Ġval Ġconj unct s Ġ= Ġmutable ListOf < String >() ĊĠĠĠ Ġif Ġ( filter . external Measurement Consumer Id Ġ!= Ġ 0 L ) Ġ{ ĊĠĠĠĠĠ Ġconj unct s . add (" External Measurement Consumer Id Ġ= Ġ@$ EXTERNAL _ MEAS UREMENT _ CON SUM ER _ ID _ PARAM ") ĊĠĠĠĠĠ Ġbind ( EXTERNAL _ MEAS UREMENT _ CON SUM ER _ ID _ PARAM Ġto Ġfilter . external Measurement Consumer Id ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( filter . external Measurement Consumer Certificate Id Ġ!= Ġ 0 L ) Ġ{ ĊĠĠĠĠĠ Ġconj unct s . add ( ĊĠĠĠĠĠĠĠ Ġ" External Measurement Consumer Certificate Id Ġ= Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠ Ġ"@ $ EXTERNAL _ MEAS UREMENT _ CON SUM ER _ CER TIFICATE _ ID _ PARAM " ĊĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġbind ( ĊĠĠĠĠĠĠĠ ĠEXTER NAL _ MEAS UREMENT _ CON SUM ER _ CER TIFICATE _ ID _ PARAM Ġto ĊĠĠĠĠĠĠĠĠĠ Ġfilter . external Measurement Consumer Certificate Id ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( filter . states Value List . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġconj unct s . add (" Measure ments . State ĠIN ĠUN NE ST (@ $ STAT ES _ PARAM )") ĊĠĠĠĠĠ Ġbind ( STAT ES _ PARAM ). toInt 6 4 Array ( filter . states Value List . map Ġ{ Ġit . to Long () Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( filter . has Updated After ()) Ġ{ ĊĠĠĠĠĠ Ġif Ġ( filter . has External Measurement Id After ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġconj unct s . add ( ĊĠĠĠĠĠĠĠĠĠ Ġ""" ĊĠĠĠĠĠĠĠĠĠ Ġ(( UpdateTime Ġ> Ġ@$ UPDATED _ AFTER ) ĊĠĠĠĠĠĠĠĠĠ ĠOR Ġ( UpdateTime Ġ= Ġ@$ UPDATED _ AFTER ĊĠĠĠĠĠĠĠĠĠ ĠAND ĠExternal Measurement Id Ġ> Ġ@$ EXTERNAL _ MEAS UREMENT _ ID _ AFTER )) ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġbind ( EXTERNAL _ MEAS UREMENT _ ID _ AFTER ). to ( filter . external Measurement Id After ) ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( filter . has External Computation Id After ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġconj unct s . add ( ĊĠĠĠĠĠĠĠĠĠ Ġ""" ĊĠĠĠĠĠĠĠĠĠ Ġ(( UpdateTime Ġ> Ġ@$ UPDATED _ AFTER ) ĊĠĠĠĠĠĠĠĠĠ ĠOR Ġ( UpdateTime Ġ= Ġ@$ UPDATED _ AFTER ĊĠĠĠĠĠĠĠĠĠ ĠAND ĠExternal Computation Id Ġ> Ġ@$ EXTERNAL _ COMP UT ATION _ ID _ AFTER )) ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġbind ( EXTERNAL _ COMP UT ATION _ ID _ AFTER ). to ( filter . external Computation Id After ) ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġerror (" external _ measurement _ id _ after Ġor Ġexternal _ measurement _ id _ after Ġrequired ") ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġbind ( UPDATED _ AFTER Ġto Ġfilter . updated After . to G cloud Timestamp ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( conj unct s . isEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġappend Clause (" WHERE Ġ") ĊĠĠĠ Ġappend ( conj unct s . join ToString (" ĠAND Ġ")) ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġconst Ġval ĠLIMIT _ PARAM Ġ= Ġ" limit " ĊĠĠĠ Ġconst Ġval ĠEXTER NAL _ MEAS UREMENT _ CON SUM ER _ ID _ PARAM Ġ= Ġ" external Measurement Consumer Id " ĊĠĠĠ Ġconst Ġval ĠEXTER NAL _ MEAS UREMENT _ CON SUM ER _ CER TIFICATE _ ID _ PARAM Ġ= ĊĠĠĠĠĠ Ġ" external Measurement Consumer Certificate Id " ĊĠĠĠ Ġconst Ġval ĠUP D ATED _ AFTER Ġ= Ġ" updated After " ĊĠĠĠ Ġconst Ġval ĠSTAT ES _ PARAM Ġ= Ġ" states " ĊĠĠĠ Ġconst Ġval ĠEXTER NAL _ MEAS UREMENT _ ID _ AFTER Ġ= Ġ" external Measurement Id After " ĊĠĠĠ Ġconst Ġval ĠEXTER NAL _ COMP UT ATION _ ID _ AFTER Ġ= Ġ" external Computation Id After " ĊĠ Ġ} Ċ } Ċ