/* Ċ Ġ* ĠCopyright Ġ@ Ġ 2 0 1 8 Ġ- Ġpresent Ġ 8 x 8 , ĠInc . Ċ Ġ* Ċ Ġ* Ġ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 . j its i . video bridge . transport . udp Ċ Ċ import Ġorg . ice 4 j . socket . Socket Closed Exception Ċ import Ġorg . j its i . nl j . util . Ordered JsonObject Ċ import Ġorg . j its i . utils . logging 2 . Logger Ċ import Ġorg . j its i . utils . logging 2 . create Child Logger Ċ import Ġorg . j its i . utils . stats . Rate Statistics Ċ import Ġjava . io . IOException Ċ import Ġjava . net . Data gram Packet Ċ import Ġjava . net . Data gram Socket Ċ import Ġjava . net . Inet Address Ċ import Ġjava . net . Inet SocketAddress Ċ import Ġjava . net . SocketAddress Ċ import Ġjava . net . Socket Exception Ċ import Ġjava . net . Unknown Host Exception Ċ import Ġjava . time . Clock Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjava . util . concurrent . atomic . Long Ad der Ċ Ċ /** Ċ Ġ* ĠA Ġtransport Ġwhich Ġallows Ġsending Ġand Ġreceiving Ġdata Ġover ĠUDP . Ġ ĠThe Ġtransport Ċ Ġ* Ġwill Ġbind Ġto Ġ[ bind Port ] Ġon Ġ[ bind Address ] Ġand Ġwill Ġlisten Ġfor Ġincoming Ċ Ġ* Ġpackets Ġthere Ġvia Ġ[ start Reading Data ] Ġuntil Ġthe Ġtransport Ġis Ġstopped . ĠSending Ċ Ġ* Ġcan Ġbe Ġdone Ġvia Ġ[ send ], Ġand Ġa Ġremote Ġaddress Ġ( or Ġa Ġset Ġof Ġremote Ġaddresses ) Ċ Ġ* Ġmust Ġbe Ġprovided . Ċ Ġ*/ Ċ class ĠU dp Transport Ġ@ Jvm Over loads Ġ@ Throws ( Socket Exception :: class , ĠUnknown Host Exception :: class ) Ġconstructor ( ĊĠĠĠ Ġprivate Ġval Ġbind Address : ĠString , ĊĠĠĠ Ġprivate Ġval Ġbind Port : ĠInt , ĊĠĠĠ Ġparent Logger : ĠLogger , ĊĠĠĠ Ġso R cv Buf : ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġso S nd Buf : ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġclock : ĠClock Ġ= ĠClock . system UTC () Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= Ġcreate Child Logger ( parent Logger , Ġmap Of ( ĊĠĠĠĠĠĠĠ Ġ" address " Ġto Ġbind Address , ĊĠĠĠĠĠĠĠ Ġ" port " Ġto Ġbind Port . toString () ĊĠĠĠ Ġ)) ĊĊĠĠĠ Ġprivate Ġval Ġrunning Ġ= ĠAtomic Boolean ( true ) ĊĊĠĠĠ Ġprivate Ġval Ġsocket : ĠData gram Socket Ġ= ĠData gram Socket ( ĊĠĠĠĠĠĠĠ ĠInet SocketAddress ( Inet Address . get ByName ( bind Address ), Ġbind Port ) ĊĠĠĠ Ġ). apply Ġ{ ĊĠĠĠĠĠĠĠ Ġso R cv Buf ?. let Ġ{ Ġreceive BufferSize Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġso S nd Buf ?. let Ġ{ Ġsend BufferSize Ġ= Ġit Ġ} ĊĠĠĠ Ġ}. also Ġ{ Ġsocket Ġ-> ĊĠĠĠĠĠĠĠ Ġlogger . info (" Initialized Ġwith Ġbind Ġaddress Ġ$ bind Address Ġand Ġbind Ġport Ġ$ bind Port . Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Receive Ġbuffer Ġsize Ġ${ socket . receive BufferSize }${ so R cv Buf ?. let Ġ{ Ġ" Ġ( ask ed Ġfor Ġ$ it ) "} Ġ?: Ġ"" }. Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Send Ġbuffer Ġsize Ġ${ socket . send BufferSize }${ so S nd Buf ?. let Ġ{ Ġ" Ġ( ask ed Ġfor Ġ$ it ) "} Ġ?: Ġ"" }. ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġstats Ġ= ĠStats () ĊĊĠĠĠ Ġvar Ġincoming Data Handler : ĠIncoming Data Handler ? Ġ= Ġnull ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠRead Ġdata Ġfor Ġas Ġlong Ġas Ġthis Ġtransport Ġis Ġstill Ġrunning . Ġ ĠReceived Ġdata ĊĠĠĠĠ Ġ* Ġis Ġpassed Ġto Ġthe Ġset Ġ[ Incoming Data Handler ]. ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġstart Reading Data () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuf Ġ= ĠByteArray ( 1 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġpacket Ġ= ĠData gram Packet ( buf , Ġ 0 , Ġ 1 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġwhile Ġ( running . get ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO : Ġneeded ? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpacket . setData ( buf , Ġ 0 , Ġ 1 5 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . receive ( packet ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( s ce : ĠSocket Closed Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Socket Ġclosed , Ġstopping Ġreader ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Exception Ġwhile Ġreading Ġ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnow Ġ= Ġclock . instant () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstats . packet Received ( packet . length , Ġnow ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġincoming Data Handler ?. data Received ( buf , Ġpacket . offset , Ġpacket . length , Ġnow ) Ġ?: Ġstats . incoming Packet Dropped () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠSend Ġdata Ġout Ġvia Ġthis Ġtransport Ġto Ġ[ remote Address ]. ĠDoes Ġnot Ġtake Ġownership ĊĠĠĠĠ Ġ* Ġof Ġthe Ġgiven Ġbuffer . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsend ( data : ĠByteArray , Ġoff : ĠInt , Ġlength : ĠInt , Ġremote Address : ĠSocket Address ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! running . get ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstats . outgoing Packet Dropped () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . send ( Data gram Packet ( data , Ġoff , Ġlength , Ġremote Address ). apply Ġ{ Ġsocket Address Ġ= Ġremote Address Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstats . packet Sent ( length , Ġclock . instant ()) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Error Ġsending Ġdata ", Ġt ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠSend Ġdata Ġout Ġvia Ġthis Ġtransport Ġto Ġ[ remote Addresses ]. ĠDoes Ġnot Ġtake Ġownership ĊĠĠĠĠ Ġ* Ġof Ġthe Ġgiven Ġbuffer . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsend ( data : ĠByteArray , Ġoff : ĠInt , Ġlength : ĠInt , Ġremote Addresses : ĠSet < SocketAddress >) Ġ{ ĊĠĠĠĠĠĠĠ Ġremote Addresses . forEach Ġ{ Ġsend ( data , Ġoff , Ġlength , Ġit ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠStop Ġthis Ġtransport . Ġ ĠIt Ġwill Ġstop Ġreceiving Ġfrom Ġthe Ġsocket Ġ( and Ġclose ĊĠĠĠĠ Ġ* Ġit ) Ġand Ġwill Ġno Ġlonger Ġsend Ġdata ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( running . compare AndSet ( true , Ġfalse )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . close () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget Stats (): ĠStats Snapshot Ġ= Ġstats . to Snapshot () ĊĊĠĠĠ Ġclass ĠStats Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġpackets Received Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġbytes Received Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġincoming Packets Dropped Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġpackets Sent Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġbytes Sent Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġoutgoing Packets Dropped Ġ= ĠLong Ad der () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġreceive Packet Rate : ĠRate Statistics Ġ= ĠRate Statistics ( RATE _ INTERVAL , ĠSC ALE ) ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġreceive Bit Rate : ĠRate Statistics Ġ= ĠRate Statistics ( RATE _ INTERVAL ) ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsend Packet Rate : ĠRate Statistics Ġ= ĠRate Statistics ( RATE _ INTERVAL , ĠSC ALE ) ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsend Bit Rate : ĠRate Statistics Ġ= ĠRate Statistics ( RATE _ INTERVAL ) ĊĊĠĠĠĠĠĠĠ Ġfun Ġpacket Received ( num Bytes : ĠInt , Ġtime : ĠInstant ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpackets Received . increment () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbytes Received . add ( num Bytes . to Long ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtime . to Epoch Milli (). also Ġ{ Ġtime Ms Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceive Packet Rate . update ( 1 , Ġtime Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceive Bit Rate . update ( num Bytes , Ġtime Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġpacket Sent ( num Bytes : ĠInt , Ġtime : ĠInstant ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpackets Sent . increment () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbytes Sent . add ( num Bytes . to Long ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtime . to Epoch Milli (). also Ġ{ Ġtime Ms Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Packet Rate . update ( 1 , Ġtime Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bit Rate . update ( num Bytes , Ġtime Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġincoming Packet Dropped () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġincoming Packets Dropped . increment () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġoutgoing Packet Dropped () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoutgoing Packets Dropped . increment () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun ĠtoJson (): ĠOrdered JsonObject Ġ= ĠOrdered JsonObject (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" packets _ received ", Ġpackets Received . sum ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" receive _ packet _ rate _ pp s ", Ġreceive Packet Rate . rate ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" incoming _ packets _ d ropped ", Ġincoming Packets Dropped . sum ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" bytes _ received ", Ġbytes Received . sum ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" packets _ sent ", Ġpackets Sent . sum ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" send _ packet _ rate _ pp s ", Ġsend Packet Rate . rate ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" outgoing _ packets _ d ropped ", Ġoutgoing Packets Dropped . sum ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" bytes _ sent ", Ġbytes Sent . sum ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġto Snapshot (): ĠStats Snapshot Ġ= ĠStats Snapshot ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpackets Received Ġ= Ġpackets Received . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbytes Received Ġ= Ġbytes Received . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġincoming Packets Dropped Ġ= Ġincoming Packets Dropped . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpackets Sent Ġ= Ġpackets Sent . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbytes Sent Ġ= Ġbytes Sent . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoutgoing Packets Dropped Ġ= Ġoutgoing Packets Dropped . sum (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreceive Packet Rate Ġ= Ġreceive Packet Rate . rate , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreceive Bit Rate Ġ= Ġreceive Bit Rate . rate , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Packet Rate Ġ= Ġsend Packet Rate . rate , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bit Rate Ġ= Ġsend Bit Rate . rate ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconst Ġval ĠR ATE _ INTERVAL : ĠInt Ġ= Ġ 6 0 0 0 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconst Ġval ĠSC ALE : ĠFloat Ġ= Ġ 1 0 0 0 f ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdata Ġclass ĠStats Snapshot ( ĊĠĠĠĠĠĠĠ Ġval Ġpackets Received : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġbytes Received : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġincoming Packets Dropped : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġpackets Sent : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġbytes Sent : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġoutgoing Packets Dropped : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġreceive Packet Rate : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġreceive Bit Rate : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġsend Packet Rate : ĠLong , ĊĠĠĠĠĠĠĠ Ġval Ġsend Bit Rate : ĠLong ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġinterface ĠIncoming Data Handler Ġ{ ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠNotify Ġthe Ġhandler Ġthat Ġdata Ġwas Ġreceived Ġ( contained ĊĠĠĠĠĠĠĠĠ Ġ* Ġwithin Ġ[ data ] Ġat Ġ[ offset ] Ġwith Ġ[ length ]) Ġat Ġ[ received Time ]) ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* ĠNote Ġthat Ġthe Ġhandler Ġdoes Ġ* not * Ġown Ġthe Ġbuffer , Ġand Ġmust Ġcopy Ġif Ġthey ĊĠĠĠĠĠĠĠĠ Ġ* Ġwant Ġto Ġmodify Ġit Ġor Ġkeep Ġit Ġlonger Ġthan Ġthe Ġduration Ġof Ġthe ĊĠĠĠĠĠĠĠĠ Ġ* Ġ[ data Received ] Ġcall . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġfun Ġdata Received ( data : ĠByteArray , Ġoffset : ĠInt , Ġlength : ĠInt , Ġreceived Time : ĠInstant ) ĊĠĠĠ Ġ} Ċ } Ċ