udp-streaming (empty) → 0.1.0.0

raw patch · 3 files changed

+175/−0 lines, 3 filesdep +basedep +bytestringdep +network

Dependencies added: base, bytestring, network, resourcet, streaming

Files

• LICENSE +21/−0
• src/Streaming/UDP.hs +120/−0
• udp-streaming.cabal +34/−0

@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2018 Mihai Giurgeanu++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.

@@ -0,0 +1,120 @@+{-# LANGUAGE OverloadedStrings #-}++{-|+Module          : Streaming.UDP+Description     : Simple fire-and-forget UDP components for Streaming library+Copyright       : (c) 2018 Mihai Giurgeanu+Stability       : experimental+Portability     : GHC+-}++module Streaming.UDP+  ( fromUDP,+    fromSocket,+    fromIOSocket,+    toUDP,+    toSocket,+    toIOSocket,+  ) where++import           Streaming (Of, Stream, lift)+import qualified Streaming.Prelude as S+import           Network.Socket (Socket,+                                 SockAddr(SockAddrInet),+                                 PortNumber,+                                 setSocketOption)+import qualified Network.Socket as Socket+import           Network.Socket.ByteString (recvFrom, send)+import           Control.Monad.IO.Class (MonadIO, liftIO)+import           Control.Monad.Trans.Resource (MonadResource, allocate)+import           Data.ByteString (ByteString)+++-- | gets an ip adress, a port and stream of strict `ByteString`s and+-- sends all the `ByteString`s to UDP socket+toUDP :: (MonadResource m) => String -> PortNumber -> Stream (Of ByteString) m r -> m r+toUDP host port = toIOSocket (udpSocket host port SinkSocket)++-- | Stream all incoming data to the given connected 'Socket'. Note that this function will+-- not automatically close the 'Socket' when processing completes.+toSocket :: (MonadIO m) => Socket -> Stream (Of ByteString) m r -> m r+toSocket s = loop . S.next+  where+    loop :: (MonadIO m) => m (Either r (ByteString, Stream (Of ByteString) m r)) -> m r+    loop m = m >>= either return loop'+    loop' :: MonadIO m => (ByteString, Stream (Of ByteString) m r) -> m r+    loop' (msg, str') = liftIO (send s msg) >> (loop $ S.next str')+                        -- since we are sending a datagram, can't use sendAll; ignore the value returned by send++-- | An alternative to 'toSocket'. Instead of taking a pre-opened 'Socket', it+-- takes an action that returns a connected 'Socket' , so that it can open it+-- only when needed and close it as soon as possible.+toIOSocket :: (MonadResource m) => IO Socket -> Stream (Of ByteString) m r -> m r+toIOSocket connectAction str = do+  (_, s) <- allocate connectAction Socket.close+  toSocket s str+++-- | Fire-and-forget style UDP source. It will attempt to listen on the+-- specified interface and port, and if it succeeds it can read contents+-- being sent to that interface and port.+-- It streams a pair of (message, source-address) allowing the downstream+-- to know who sent the message. +fromUDP :: (MonadResource m)+        => String               -- ^ address to bind to+        -> PortNumber           -- ^ port number to bind to+        -> Int                  -- ^ maximum size of a datagram+        -> Stream (Of (ByteString, SockAddr)) m r+fromUDP host port sz =+  fromIOSocket (udpSocket host port SourceSocket) sz++-- | Stream the contents of a bound 'Socket' as binary data. Note that this function will+-- not automatically close the 'Socket' when processing completes, since it did not acquire+-- the 'Socket' in the first place.+fromSocket :: (MonadIO m)+           => Socket                                 -- ^ the socket+           -> Int                                    -- ^ max lenght of a datagram+           -> Stream (Of (ByteString, SockAddr)) m r -- ^ the resulting stream of messages+fromSocket s sz = loop+  where loop = do+          msg <- liftIO $ recvFrom s sz+          S.yield msg+          loop++-- | An alternative to 'fromSocket'. Instead of taking a pre-bound 'Socket',+-- it takes an action that opens and binds a Socket, so that it can open it+-- only when needed and close it as soon as possible.+fromIOSocket :: (MonadResource m)+             => IO Socket                              -- ^ the 'IO' action to create the UDP socket+             -> Int                                    -- ^ max length of a datagram+             -> Stream (Of (ByteString, SockAddr)) m r -- ^ the resulting stream of messages+fromIOSocket bindAction sz = do+  (_, s) <- lift $ allocate bindAction Socket.close+  fromSocket s sz++-- | Given a host, port and 'SocketType', create a socket and+-- 'Network.Socket.connect' or 'Network.Socket.bind' to the address pair.+udpSocket+    :: String -- ^ IP address to host+    -> PortNumber+    -> SocketType+    -> IO Socket+udpSocket host port socketType =  do+    socket <- Socket.socket Socket.AF_INET Socket.Datagram Socket.defaultProtocol+    setSocketOption socket Socket.ReusePort 1+    address <- Socket.inet_addr host+    socketMode socketType socket (SockAddrInet port address)+    return socket++-- | A type to distinguish when we want to read and write from sockets,+-- simplifying the implementation+data SocketType+    = SourceSocket -- ^ source socket, should 'Network.Socket.bind'+    | SinkSocket   -- ^ sink socket, should 'Network.Socket.connect'++-- | Use either 'Network.Socket.bind' or 'Network.Socket.connect' depending on the 'SocketType'+socketMode :: SocketType -> Socket -> SockAddr -> IO ()+socketMode socketType =+    case socketType of+        SourceSocket -> Socket.bind+        SinkSocket -> Socket.connect

@@ -0,0 +1,34 @@+-- This file has been generated from package.yaml by hpack version 0.20.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 7e911d679bae02bfce07a28bc4e8d4e264414212bd4957071ee7f7b7779de1c4++name:           udp-streaming+version:        0.1.0.0+synopsis:       Simple fire-and-forget udp Streaming components modelled after udp-conduit+category:       Network, Streaming+stability:      experimental+homepage:       https://hub.darcs.net/mihaigiurgeanu/udp-streaming+author:         Mihai Giurgeanu+maintainer:     mihai.giurgeanu@gmail.com+copyright:      (c) 2018 Mihai Giurgeanu+license:        MIT+license-file:   LICENSE+build-type:     Simple+cabal-version:  >= 1.10++library+  hs-source-dirs:+      src+  build-depends:+      base >=4.6 && <5.0+    , bytestring >=0.10 && <0.11+    , network >=2.0 && <3+    , resourcet >=1.2 && <2+    , streaming >=0.2 && <0.3+  exposed-modules:+      Streaming.UDP+  other-modules:+      Paths_udp_streaming+  default-language: Haskell2010