-- | Tests for the UDP example event sink.
module Hydra.Events.UDPSpec where

import Hydra.Prelude
import Test.Hydra.Prelude

import Control.Monad.Trans.Resource (runResourceT)
import Data.ByteString.Char8 qualified as BS8
import Hydra.Events (EventId, EventSink (..), putEvent)
import Hydra.Events.UDP (newUDPEventSink, withUDPEventSink)
import Network.Socket (PortNumber, socketPort)
import Network.UDP (ListenSocket (..), recvFrom, serverSocket, stop)

spec :: Spec
spec :: Spec
spec = do
  ServiceName -> Spec -> Spec
forall a. HasCallStack => ServiceName -> SpecWith a -> SpecWith a
describe ServiceName
"putEvent" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    ServiceName -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
ServiceName -> a -> SpecWith (Arg a)
it ServiceName
"sends datagram" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a. (ListenSocket -> PortNumber -> IO a) -> IO a
withUDPServerSocket ((ListenSocket -> PortNumber -> IO ()) -> IO ())
-> (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ListenSocket
socket PortNumber
port -> do
        ServiceName
-> ServiceName -> (EventSink EventId IO -> IO ()) -> IO ()
forall e a.
ToJSON e =>
ServiceName -> ServiceName -> (EventSink e IO -> IO a) -> IO a
withUDPEventSink ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port) ((EventSink EventId IO -> IO ()) -> IO ())
-> (EventSink EventId IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventSink EventId IO
sink -> do
          let event :: EventId
event = EventId
123 :: EventId
          EventSink EventId IO -> HasEventId EventId => EventId -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink EventId IO
sink EventId
event
          (ByteString
received, ClientSockAddr
_) <- ListenSocket -> IO (ByteString, ClientSockAddr)
recvFrom ListenSocket
socket
          ByteString -> ServiceName
BS8.unpack ByteString
received ServiceName -> ServiceName -> IO ()
forall a. (HasCallStack, Show a, Eq a) => [a] -> [a] -> IO ()
`shouldContain` ServiceName
"123"

    ServiceName -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
ServiceName -> a -> SpecWith (Arg a)
it ServiceName
"allows concurrent usage" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a. (ListenSocket -> PortNumber -> IO a) -> IO a
withUDPServerSocket ((ListenSocket -> PortNumber -> IO ()) -> IO ())
-> (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ListenSocket
_ PortNumber
port -> do
        forall e a.
ToJSON e =>
ServiceName -> ServiceName -> (EventSink e IO -> IO a) -> IO a
withUDPEventSink @EventId ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port) ((EventSink EventId IO -> IO ()) -> IO ())
-> (EventSink EventId IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventSink{HasEventId EventId => EventId -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId EventId => EventId -> IO ()
putEvent} -> do
          (ServiceName, IO ()) -> (ServiceName, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(ServiceName, m a) -> (ServiceName, m b) -> m ()
concurrentlyLabelled_ (ServiceName
"put-event-123", EventId -> IO ()
HasEventId EventId => EventId -> IO ()
putEvent EventId
123) (ServiceName
"put-event-456", EventId -> IO ()
HasEventId EventId => EventId -> IO ()
putEvent EventId
456)

  ServiceName -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
ServiceName -> a -> SpecWith (Arg a)
it ServiceName
"supports multiple instances" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
    (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a. (ListenSocket -> PortNumber -> IO a) -> IO a
withUDPServerSocket ((ListenSocket -> PortNumber -> IO ()) -> IO ())
-> (ListenSocket -> PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ListenSocket
_ PortNumber
port -> do
      forall e a.
ToJSON e =>
ServiceName -> ServiceName -> (EventSink e IO -> IO a) -> IO a
withUDPEventSink @EventId ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port) ((EventSink EventId IO -> IO ()) -> IO ())
-> (EventSink EventId IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventSink EventId IO
s1 -> do
        forall e a.
ToJSON e =>
ServiceName -> ServiceName -> (EventSink e IO -> IO a) -> IO a
withUDPEventSink @EventId ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port) ((EventSink EventId IO -> IO ()) -> IO ())
-> (EventSink EventId IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventSink EventId IO
s2 -> do
          EventSink EventId IO -> HasEventId EventId => EventId -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink EventId IO
s1 EventId
123
          EventSink EventId IO -> HasEventId EventId => EventId -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink EventId IO
s2 EventId
456

      ResourceT IO () -> IO ()
forall (m :: * -> *) a. MonadUnliftIO m => ResourceT m a -> m a
runResourceT (ResourceT IO () -> IO ()) -> ResourceT IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        EventSink EventId IO
s1 <- forall e (m :: * -> *).
(ToJSON e, MonadResource m) =>
ServiceName -> ServiceName -> m (EventSink e IO)
newUDPEventSink @EventId ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port)
        EventSink EventId IO
s2 <- forall e (m :: * -> *).
(ToJSON e, MonadResource m) =>
ServiceName -> ServiceName -> m (EventSink e IO)
newUDPEventSink @EventId ServiceName
"0.0.0.0" (PortNumber -> ServiceName
forall b a. (Show a, IsString b) => a -> b
show PortNumber
port)
        IO () -> ResourceT IO ()
forall (m :: * -> *) a. Monad m => m a -> ResourceT m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO () -> ResourceT IO ()) -> IO () -> ResourceT IO ()
forall a b. (a -> b) -> a -> b
$ EventSink EventId IO -> HasEventId EventId => EventId -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink EventId IO
s1 EventId
123
        IO () -> ResourceT IO ()
forall (m :: * -> *) a. Monad m => m a -> ResourceT m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO () -> ResourceT IO ()) -> IO () -> ResourceT IO ()
forall a b. (a -> b) -> a -> b
$ EventSink EventId IO -> HasEventId EventId => EventId -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink EventId IO
s2 EventId
456

-- | Bind a UDP server socket on any free port and hand the assigned port to
-- the action. Binding port 0 directly avoids probing for a free port with a
-- separate socket, which is racy (and the previous TCP probe said nothing
-- about UDP availability anyway).
withUDPServerSocket :: (ListenSocket -> PortNumber -> IO a) -> IO a
withUDPServerSocket :: forall a. (ListenSocket -> PortNumber -> IO a) -> IO a
withUDPServerSocket ListenSocket -> PortNumber -> IO a
action =
  IO ListenSocket
-> (ListenSocket -> IO ()) -> (ListenSocket -> IO a) -> IO a
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket ((IP, PortNumber) -> IO ListenSocket
serverSocket (IP
"0.0.0.0", PortNumber
0)) ListenSocket -> IO ()
stop ((ListenSocket -> IO a) -> IO a) -> (ListenSocket -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \ListenSocket
socket -> do
    PortNumber
port <- Socket -> IO PortNumber
socketPort (ListenSocket -> Socket
listenSocket ListenSocket
socket)
    ListenSocket -> PortNumber -> IO a
action ListenSocket
socket PortNumber
port