/* Ċ Ġ* ĠCopyright Ġ 2 0 1 4 - 2 0 2 0 ĠJetBrains Ġs . r . o Ġand Ġcontributors . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ package Ġio . ktor . network . sockets Ċ Ċ import Ġio . ktor . network . selector .* Ċ import Ġio . ktor . network . util .* Ċ import Ġio . ktor . util . network .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . channels .* Ċ import Ġio . ktor . utils . io . core .* Ċ import Ġjava . net .* Ċ import Ġjava . nio .* Ċ import Ġjava . nio . channels .* Ċ Ċ @ Suppress (" Blocking Method In Non Blocking Context ") Ċ internal Ġclass ĠData gram Socket Impl ( override Ġval Ġchannel : ĠData gram Channel , Ġselector : ĠSelector Manager ) ĊĠĠĠ Ġ: ĠBound Data gram Socket , ĠConnected Data gram Socket , ĠN IO Socket Impl < Data gram Channel >( channel , Ġselector , ĠDefault Data gram ByteBuffer Pool ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġsocket Ġ= Ġchannel . socket () !! ĊĊĠĠĠ Ġoverride Ġval Ġlocal Address : ĠNetwork Address ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġsocket . local SocketAddress Ġas ? ĠNetwork Address ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠIllegalStateException (" Channel Ġis Ġnot Ġyet Ġbound ") ĊĊĠĠĠ Ġoverride Ġval Ġremote Address : ĠNetwork Address ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġsocket . remote SocketAddress Ġas ? ĠNetwork Address ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠIllegalStateException (" Channel Ġis Ġnot Ġyet Ġconnected ") ĊĊĠĠĠ Ġ@ Opt In ( Obsolete Cor outines Api :: class , ĠExperimental Cor outines Api :: class ) ĊĠĠĠ Ġprivate Ġval Ġsender Ġ= Ġactor < Data gram >( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠ Ġconsume Each Ġ{ Ġdata gram Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Impl ( data gram ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Opt In ( Experimental Cor outines Api :: class ) ĊĠĠĠ Ġprivate Ġval Ġreceiver Ġ= Ġproduce < Data gram >( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . send ( receive Impl ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġval Ġincoming : ĠReceive Channel < Data gram > ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġreceiver ĊĊĠĠĠ Ġoverride Ġval Ġoutgoing : ĠSend Channel < Data gram > ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġsender ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġreceiver . cancel () ĊĠĠĠĠĠĠĠ Ġsender . close () ĊĠĠĠĠĠĠĠ Ġsuper . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġreceive Impl (): ĠData gram Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= ĠDefault Data gram ByteBuffer Pool . borrow () ĊĠĠĠĠĠĠĠ Ġval Ġaddress Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . receive ( buffer ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠDefault Data gram ByteBuffer Pool . recycle ( buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġt ĊĠĠĠĠĠĠĠ Ġ} Ġ?: Ġreturn Ġreceive Suspend ( buffer ) ĊĊĠĠĠĠĠĠĠ Ġinterest Op ( Select Interest . READ , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġbuffer . flip () ĊĠĠĠĠĠĠĠ Ġval Ġdata gram Ġ= ĠData gram ( build Packet Ġ{ Ġwrite Fully ( buffer ) Ġ}, Ġaddress ) ĊĠĠĠĠĠĠĠ ĠDefault Data gram ByteBuffer Pool . recycle ( buffer ) ĊĠĠĠĠĠĠĠ Ġreturn Ġdata gram ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġreceive Suspend ( buffer : ĠByteBuffer ): ĠData gram Ġ{ ĊĠĠĠĠĠĠĠ Ġinterest Op ( Select Interest . READ , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġselector . select ( this , ĠSelect Interest . READ ) ĊĊĠĠĠĠĠĠĠ Ġval Ġaddress Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . receive ( buffer ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠDefault Data gram ByteBuffer Pool . recycle ( buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġt ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( address Ġ== Ġnull ) Ġreturn Ġreceive Suspend ( buffer ) ĊĊĠĠĠĠĠĠĠ Ġinterest Op ( Select Interest . READ , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġbuffer . flip () ĊĠĠĠĠĠĠĠ Ġval Ġdata gram Ġ= ĠData gram ( build Packet Ġ{ Ġwrite Fully ( buffer ) Ġ}, Ġaddress ) ĊĠĠĠĠĠĠĠ ĠDefault Data gram ByteBuffer Pool . recycle ( buffer ) ĊĠĠĠĠĠĠĠ Ġreturn Ġdata gram ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġsend Impl ( data gram : ĠData gram ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= ĠByteBuffer . allocate Direct ( data gram . packet . remaining . toInt ()) ĊĠĠĠĠĠĠĠ Ġdata gram . packet . read Available ( buffer ) ĊĠĠĠĠĠĠĠ Ġbuffer . flip () ĊĊĠĠĠĠĠĠĠ Ġval Ġrc Ġ= Ġchannel . send ( buffer , Ġdata gram . address ) ĊĠĠĠĠĠĠĠ Ġif Ġ( rc Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Suspend ( buffer , Ġdata gram . address ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinterest Op ( Select Interest . WRITE , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġsend Suspend ( buffer : ĠByteBuffer , Ġaddress : ĠSocket Address ) Ġ{ ĊĠĠĠĠĠĠĠ Ġinterest Op ( Select Interest . WRITE , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġselector . select ( this , ĠSelect Interest . WRITE ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ( channel . send ( buffer , Ġaddress ) Ġ== Ġ 0 ) Ġsend Suspend ( buffer , Ġaddress ) ĊĠĠĠĠĠĠĠ Ġelse Ġinterest Op ( Select Interest . WRITE , Ġfalse ) ĊĠĠĠ Ġ} Ċ } Ċ