/* Ċ Ġ* Ġ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 . functions Ċ Ċ import Ġorg . apache . beam . sdk . transforms . Wait Ċ import Ġorg . apache . beam . sdk . transforms . window ing .* Ċ import Ġorg . apache . beam . sdk . values .* Ċ import Ġorg . joda . time . Duration Ċ import Ġru . ch erm en in . k io . utils . Configurator Ċ import Ġru . ch erm en in . k io . utils . hash WithName Ċ Ċ /** Ċ Ġ* ĠMethod Ġto Ġwindow Ġcollection Ġby Ġdefined Ġfunction . Ċ Ġ* Ċ Ġ* Ġ@ param Ġwindow Fn Ġwindow Ġfunction Ċ Ġ* Ġ@ param Ġoptions Ġadditional Ġwindow ing Ġoptions Ġ( optional ) Ċ Ġ* Ġ@ return Ġwindow ed Ġcollection Ċ Ġ*/ Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Window ( ĊĠĠĠ Ġwindow Fn : ĠWindow Fn < T , Ġ* >, ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Ġ= Ġoptions . invoke ( Window . into ( window Fn )) ĊĠĠĠ Ġreturn Ġthis . apply ( window . hash WithName (" with Window "), Ġwindow ) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Window By Days ( ĊĠĠĠ Ġnumber : ĠInt , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠCalendar Windows . days ( number ) Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Calendar Window By Days ($ number ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Window By Week s ( ĊĠĠĠ Ġnumber : ĠInt , ĊĠĠĠ Ġstart DayOfWeek : ĠInt , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠCalendar Windows . week s ( number , Ġstart DayOfWeek ) Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Calendar Window By Week s ($ number , Ġ$ start DayOfWeek ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Window By Months ( ĊĠĠĠ Ġnumber : ĠInt , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠCalendar Windows . months ( number ) Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Calendar Window By Months ($ number ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Window By Years ( ĊĠĠĠ Ġnumber : ĠInt , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠCalendar Windows . years ( number ) Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Calendar Window By Years ($ number ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Fixed Window ( ĊĠĠĠ Ġduration : ĠDuration , ĊĠĠĠ Ġoffset : ĠDuration Ġ= ĠDuration . ZERO , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠFixed Windows ĊĠĠĠĠĠĠĠ Ġ. of ( duration ) ĊĠĠĠĠĠĠĠ Ġ. with Offset ( offset ) ĊĠĠĠĠĠĠĠ Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Fixed Window ($ duration , Ġ$ offset ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Global Window ( ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠGlobal Windows () Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Global Window () ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Session Window ( ĊĠĠĠ Ġgap : ĠDuration , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠS essions ĊĠĠĠĠĠĠĠ Ġ. with Gap Duration ( gap ) ĊĠĠĠĠĠĠĠ Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Session Windows ($ gap ) ")) Ċ } Ċ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ inline Ġfun Ġ< T > ĠP Collection < T >. with Sl iding Window ( ĊĠĠĠ Ġsize : ĠDuration , ĊĠĠĠ Ġperiod : ĠDuration , ĊĠĠĠ Ġoffset : ĠDuration Ġ= ĠDuration . ZERO , ĊĠĠĠ Ġoptions : ĠConfigurator < Window < T >> Ġ= Ġ{ Ġit Ġ} Ċ ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwindow Fn Ġ= ĠSl iding Windows ĊĠĠĠĠĠĠĠ Ġ. of ( size ) ĊĠĠĠĠĠĠĠ Ġ. every ( period ) ĊĠĠĠĠĠĠĠ Ġ. with Offset ( offset ) ĊĠĠĠĠĠĠĠ Ġas ĠWindow Fn < T , Ġ*> ĊĠĠĠ Ġreturn Ġthis . with Window ( window Fn , Ġoptions ) ĊĠĠĠĠĠĠĠ Ġ. setName ( window Fn . hash WithName (" with Sl iding Window ($ size , Ġ$ period , Ġ$ offset ) ")) Ċ } Ċ Ċ inline Ġfun Ġ< T > ĠP Collection < T >. wait ( var arg Ġsignals : ĠP Collection < *> ): ĠP Collection < T > Ġ{ ĊĠĠĠ Ġval Ġwait er Ġ= ĠWait . on < T >(* signals ) ĊĠĠĠ Ġreturn Ġthis . apply ( wait er . hash WithName (" wait "), Ġwait er ) Ċ } Ċ