diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,48 @@ +udp-streaming +============== + +Simple fire and forget style streaming source and consumer in the spirit +of the [udp-conduit](https://github.com/kqr/udp-conduit) package. + +Here's an example run in ghci that sends e messages to UDP port 8888 on localhost, +from the port 3000 on the localhost: + + >>> import qualified Streaming.Prelude as S + >>> import Control.Monad.Trans.Resource + >>> import Streaming.UDP + >>> + >>> :set -XOverloadedStrings + >>> + >>> addr1 <- SockAddrInet 8888 <$> (inet_addr "127.0.0.1") + >>> + >>> runResourceT $ toUDP "127.0.0.1" 3000 $ S.zip (S.each ["hello", "world", "!"]) (S.repeat addr1) + +Here's an example in ghci where we continuously receive messages with a +maximum length of 500 bytes on port 8888 on localhost: + + >>> import qualified Streaming.Prelude as S + >>> import Streaming.UDP + >>> import Control.Monad.Trans.Resource + >>> import qualified Data.ByteString as B + >>> + >>> + >>> runResourceT $ S.stdoutLn $ S.map (map (toEnum.fromEnum) . B.unpack) $ S.map fst $ fromUDP "127.0.0.1" 8888 500 + + +Now, here's an example where we receive messages on port 8001, encode them as hex +digits (using base16-bytestring package), and then send them on to port 8002. + + >>> import qualified Streaming.Prelude as S + >>> import Streaming.UDP + >>> import Data.ByteString.Base16 (encode) + >>> import Control.Monad.Trans.Resource + >>> + >>> addr2 <- SockAddrInet 8002 <$> (inet_addr "127.0.0.1") + >>> + >>> runResourceT $ toUDP "127.0.0.1" 8002 $ S.zip (S.map encode $ S.map fst $ fromUDP "127.0.0.1" 8001 500) (S.repeat addr2) + +Or you could, in the previous example, to also log the messages receved, before processing and +send them further: + + >>> runResourceT $ toUDP "127.0.0.1" 3000 $ S.zip (S.map encode $ S.map fst $ S.print $ S.copy $ fromUDP "127.0.0.1" 8001 500) (S.repeat addr2) + diff --git a/src/Streaming/UDP.hs b/src/Streaming/UDP.hs --- a/src/Streaming/UDP.hs +++ b/src/Streaming/UDP.hs @@ -15,6 +15,10 @@ toUDP, toSocket, toIOSocket, + + -- * Reexports + inet_addr, + SockAddr(SockAddrInet) ) where import Streaming (Of, Stream, lift) @@ -22,30 +26,44 @@ import Network.Socket (Socket, SockAddr(SockAddrInet), PortNumber, - setSocketOption) + setSocketOption, + inet_addr) import qualified Network.Socket as Socket -import Network.Socket.ByteString (recvFrom, send) +import Network.Socket.ByteString (recvFrom, + sendTo) 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) +-- | gets an ip adress, a port and stream of strict pairs ('ByteString', 'SockAddr') and +-- sends all the `ByteString`s to the 'SockAddr' from a socet bounded to the give ip and port. +-- +-- /Note/: before 0.2.0.0 this function had a different specification. +-- +-- @since 0.2.0.0 +toUDP :: (MonadResource m) => String -> PortNumber -> Stream (Of (ByteString, SockAddr)) m r -> m r +toUDP host port = toIOSocket (udpSocket host port) -- | 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 = S.mapM_ $ liftIO . send s +-- +-- /Note/: before 0.2.0.0 this function had a different specification. +-- +-- @since 0.2.0.0 +toSocket :: (MonadIO m) => Socket -> Stream (Of (ByteString, SockAddr)) m r -> m r +toSocket s = S.mapM_ $ liftIO . (\ (m, a) -> sendTo s m a) -- 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 +-- +-- /Note/: before 0.2.0.0 this function had a different specification. +-- +-- @since 0.2.0.0 +toIOSocket :: (MonadResource m) => IO Socket -> Stream (Of (ByteString, SockAddr)) m r -> m r toIOSocket connectAction str = do (_, s) <- allocate connectAction Socket.close toSocket s str @@ -62,7 +80,7 @@ -> Int -- ^ maximum size of a datagram -> Stream (Of (ByteString, SockAddr)) m r fromUDP host port sz = - fromIOSocket (udpSocket host port SourceSocket) sz + fromIOSocket (udpSocket host port) 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 @@ -93,24 +111,10 @@ udpSocket :: String -- ^ IP address to host -> PortNumber - -> SocketType -> IO Socket -udpSocket host port socketType = do +udpSocket host port = 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) + address <- inet_addr host + Socket.bind 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 diff --git a/udp-streaming.cabal b/udp-streaming.cabal --- a/udp-streaming.cabal +++ b/udp-streaming.cabal @@ -1,12 +1,13 @@ --- This file has been generated from package.yaml by hpack version 0.20.0. +-- This file has been generated from package.yaml by hpack version 0.27.0. -- -- see: https://github.com/sol/hpack -- --- hash: fe0af0b6e00f4017f468ad1df7906de71eb634f5215bb20cd40c0a7364247174 +-- hash: 85a95f5ca918709a3c111fac4ca4f1a8fd0f982d84b8c410b97435a2c6b343e9 name: udp-streaming -version: 0.1.0.1 -synopsis: Simple fire-and-forget udp Streaming components modelled after udp-conduit +version: 0.2.0.0 +synopsis: Streaming to and from UDP socket +description: Simple fire-and-forget udp Streaming components modelled after udp-conduit category: Network, Streaming stability: experimental homepage: https://hub.darcs.net/mihaigiurgeanu/udp-streaming @@ -16,7 +17,10 @@ license: MIT license-file: LICENSE build-type: Simple -cabal-version: >= 1.10 +cabal-version: >= 1.18 + +extra-doc-files: + README.md library hs-source-dirs: