reflex-backend-socket (empty) → 0.2.0.0
raw patch · 11 files changed
+923/−0 lines, 11 filesdep +basedep +bytestringdep +containerssetup-changed
Dependencies added: base, bytestring, containers, lens, mtl, network, reflex, reflex-backend-socket, reflex-basic-host, semialign, semigroupoids, stm, these, witherable
Files
- ChangeLog.md +13/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- example/Client.hs +59/−0
- example/Others.hs +97/−0
- example/Server.hs +124/−0
- reflex-backend-socket.cabal +95/−0
- src/Reflex/Backend/Socket.hs +210/−0
- src/Reflex/Backend/Socket/Accept.hs +174/−0
- src/Reflex/Backend/Socket/Connect.hs +74/−0
- src/Reflex/Backend/Socket/Error.hs +45/−0
+ ChangeLog.md view
@@ -0,0 +1,13 @@+# Revision history for reflex-backend-socket++## 0.2.0.0 -- 2019-10-29 (first hackage release)++* Decoupled reflex-binary from this library. If you want incremental+ decoding, you can feed the `ByteString`s into reflex-binary+ yourself.++* Simplified, rewrote and documented almost all of the library.++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2017, Dave Laing++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Dave Laing nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example/Client.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE RecursiveDo #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}++{-|+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable+-}++module Main (main) where++import Control.Concurrent (forkIO)+import Control.Monad (void, forever)+import Control.Monad.Fix (MonadFix)+import Control.Monad.IO.Class (MonadIO(..))+import qualified Data.ByteString.Char8 as BC+import qualified Network.Socket as NS+import Reflex+import Reflex.Backend.Socket+import Reflex.Host.Basic+import Reflex.Workflow++type ClientCxt t m =+ ( Reflex t+ , MonadFix m+ , PerformEvent t m+ , PostBuild t m+ , TriggerEvent t m+ , MonadIO m+ , MonadIO (Performable m)+ )++unconnected :: ClientCxt t m => Workflow t m (Event t ())+unconnected = Workflow $ do+ (eError, eConnected) <- fanEither <$> connect (Just "127.0.0.1") "9000"+ eReportedError <-+ performEvent $+ (\e -> liftIO $ putStrLn "Error:\n" *> print e) <$>+ eError+ pure (() <$ eReportedError, connected <$> eConnected)++connected :: ClientCxt t m => NS.Socket -> Workflow t m (Event t ())+connected s = Workflow $ mdo+ (eLine, onLine) <- newTriggerEvent+ void . liftIO . forkIO . forever $ onLine . BC.pack =<< getLine++ so <- socket $ SocketConfig s 2048 eLine eQuit+ performEvent_ $ liftIO . BC.putStrLn <$> _sReceive so++ let eQuit = _sClose so+ pure (eQuit, unconnected <$ eQuit)++main :: IO ()+main = basicHostWithQuit $ do+ deQuit <- workflow unconnected+ pure (switchDyn deQuit)
+ example/Others.hs view
@@ -0,0 +1,97 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecursiveDo #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++{-|+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable+-}++module Main (main) where++import Control.Monad.IO.Class (liftIO)+import qualified Data.ByteString.Char8 as BC+import Data.Functor ((<&>), void)+import Data.Maybe (isNothing)+import Data.Witherable (catMaybes)+import qualified Network.Socket as NS+import Reflex+import Reflex.Backend.Socket+import Reflex.Host.Basic (basicHostForever, basicHostWithQuit)+import Reflex.Network (networkHold)+import System.Environment (getArgs)++-- | Connect to a remote host, and quit as soon as something happens.+connect1 :: IO ()+connect1 = basicHostWithQuit $ do+ eQuit <- connect (Just "127.0.0.1") "9000"+ let (eError, eConnect) = fanEither eQuit++ performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eConnect+ performEvent_ $ liftIO . print <$> eError++ pure (void eQuit)++-- | Listen for connections, and log them as they come in. Does+-- nothing with arriving connections, so will leak FDs.+accept1 :: IO ()+accept1 = basicHostForever $ do+ (eListenError, eAccept) <- fanEither <$> accept+ (AcceptConfig (Just "127.0.0.1") (Just "9000") 1 [(NS.ReuseAddr, 1)] never)++ eNewClient <- switchHold never $ _aAcceptSocket <$> eAccept+ eListenClosed <- switchHold never $ _aClose <$> eAccept+ eAcceptError <- switchHold never $ _aError <$> eAccept++ performEvent_ $+ (liftIO . putStrLn $ "Error starting listen socket") <$ eListenError+ performEvent_ $ (liftIO . putStrLn $ "Listen closed") <$ eListenClosed+ performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eNewClient+ performEvent_ $ liftIO . print <$> eAcceptError++ pure ()++-- | Connect to a remote host. When the connection succeeds, put the+-- @'Socket' t@ into @dSocket@, send a message, then close.+connect2 :: IO ()+connect2 = basicHostWithQuit $ mdo+ (eConnError, eConnect) <- fanEither <$> connect (Just "127.0.0.1") "9000"++ performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eConnect+ performEvent_ $ liftIO . print <$> eConnError++ dSocket :: Dynamic t (Maybe (Socket t)) <- networkHold (pure Nothing) $ leftmost+ [ eConnect <&> \s -> Just <$> socket (SocketConfig s 2048 eTx eOpen)+ , pure Nothing <$ eClosed+ ]++ eOpen <- switchHold never . fmap _sOpen . catMaybes $ updated dSocket+ let eTx = "Hi" <$ eOpen++ eClosed <- switchHold never . fmap _sClose . catMaybes $ updated dSocket+ eRx <- switchHold never . fmap _sReceive . catMaybes $ updated dSocket+ eSockError <- switchHold never . fmap _sError . catMaybes $ updated dSocket++ performEvent_ $ liftIO . BC.putStrLn <$> eRx+ performEvent_ $ liftIO . print <$> eSockError+ performEvent_ $ liftIO (putStrLn "Closed") <$ eClosed++ let eQuit = leftmost+ [ void eConnError+ , void . ffilter isNothing $ updated dSocket+ ]++ pure eQuit++main :: IO ()+main = getArgs >>= \case+ ["accept1"] -> accept1+ ["connect1"] -> connect1+ ["connect2"] -> connect2+ _ -> pure ()
+ example/Server.hs view
@@ -0,0 +1,124 @@+{-# LANGUAGE RecursiveDo #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-|+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable+-}++module Main (main) where++import Control.Lens (_1, _2, _3, view)+import Control.Monad (void)+import Control.Monad.Fix (MonadFix)+import Control.Monad.IO.Class (MonadIO(..))+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as BC+import Data.Char (isSpace)+import Data.Function ((&))+import Data.Functor ((<&>))+import Data.Map (Map)+import qualified Data.Map as Map+import qualified Network.Socket as NS+import Reflex+import Reflex.Backend.Socket+import Reflex.Host.Basic+import System.IO.Error++type ServerCxt t m =+ ( Reflex t+ , MonadFix m+ , MonadSample t m+ , PerformEvent t m+ , PostBuild t m+ , TriggerEvent t m+ , MonadIO m+ , MonadIO (Performable m)+ )++perConnection+ :: ServerCxt t m+ => Dynamic t NS.Socket+ -> Event t ByteString+ -> m (Event t ByteString, Event t (), Event t ())+perConnection dSocket eTx = mdo+ s <- sample . current $ dSocket+ so <- socket $ SocketConfig s 4096 eTx eQuit++ let+ eRx = _sReceive so+ eRxLine = fst . BC.spanEnd isSpace <$> eRx+ eQuit = void $ ffilter (== "quit") eRxLine+ eShutdown = void $ ffilter (== "shutdown") eRxLine+ eOut = ffilter (`notElem` ["quit", "shutdown"]) eRxLine++ pure (eOut, eShutdown, _sClose so)++go :: forall t m. BasicGuestConstraints t m => BasicGuest t m (Event t ())+go = mdo+ (eListenError, eAccept) <- fanEither <$> accept+ (AcceptConfig (Just "127.0.0.1") (Just "9000") 1 [(NS.ReuseAddr, 1)] eQuit)++ eNewClient <- switchHold never $ fmap fst . _aAcceptSocket <$> eAccept+ eListenClose <- switchHold never $ _aClose <$> eAccept+ eAcceptError <- switchHold never $ _aError <$> eAccept++ performEvent_ $ eListenError+ <&> liftIO . putStrLn . ("Listen error: " <>) . show+ performEvent_ $ eAcceptError+ <&> liftIO . putStrLn . ("Error accepting: " <>) . show . ioeGetErrorType++ eAdds :: Event t (Integer, NS.Socket) <- numberOccurrences eNewClient++ -- Maintain a map of all open sockets, indexed by their occurrence number.+ dSocketMap <- foldDyn ($) Map.empty . mergeWith (.) $+ [ uncurry Map.insert <$> eAdds+ , flip (foldr Map.delete) <$> eRemoves+ ]++ dmeSocketEvents+ :: Dynamic t (Map Integer (Event t ByteString, Event t (), Event t ()))+ <- list dSocketMap $ \dSocket -> perConnection dSocket eChats++ let+ eChats :: Event t ByteString+ eChats = dmeSocketEvents+ -- Pick out the "data received" events, and go from "dynamic map of+ -- events" to "dynamic of (event of map)"+ & fmap (mergeMap . fmap (view _1))++ -- listen to the latest Event+ & switchDyn++ -- Collect all the received data into one big ByteString+ & fmap (mconcat . Map.elems)++ eShutdown = dmeSocketEvents+ & fmap (mergeMap . fmap (view _2))+ & switchDyn+ & void++ eRemoves = dmeSocketEvents+ & fmap (mergeMap . fmap (view _3))+ & switchDyn+ & fmap Map.keys++ eQuitMap = void . ffilter id . updated $ Map.null <$> dSocketMap+ eQuit = leftmost+ [ void eListenError -- Couldn't open listen socket+ , eQuitMap -- Nobody is connected+ , eListenClose -- Our listen socket was closed+ , eShutdown -- Client asked us to shutdown everything+ ]+ pure eQuit++main :: IO ()+main = basicHostWithQuit go
+ reflex-backend-socket.cabal view
@@ -0,0 +1,95 @@+-- Initial reflex-backend-socket.cabal generated by cabal init. For further+-- documentation, see http://haskell.org/cabal/users-guide/++name: reflex-backend-socket+version: 0.2.0.0+synopsis: Reflex bindings for TCP sockets+description:+ <<https://raw.githubusercontent.com/qfpl/assets/master/data61-transparent-bg.png>>+ .+ reflex-backend-socket provides functions to handle sockets using+ Reflex @Event@s. Sending\/receiving\/waiting\/accepting are all+ performed on background threads.+ .+ The most important function in this library is+ @Reflex.Backend.Socket.socket@, which wraps a @Socket@ to process+ @Event t ByteString@s.+ .+ That @Socket@ can come from:+ .+ 1. @Reflex.Backend.Socket.Accept.accept@, if you're making a server;+ 2. @Reflex.Backend.Socket.Connect.connect@, if you're making a client; or+ 3. Your favourite networking library.+license: BSD3+license-file: LICENSE+author: Dave Laing+maintainer: dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+copyright: (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+category: Network+build-type: Simple+extra-source-files: ChangeLog.md+cabal-version: >=1.10+tested-with: GHC == 8.6.5++source-repository head+ type: git+ location: git@github.com/qfpl/reflex-backend-socket.git++library+ exposed-modules: Reflex.Backend.Socket+ , Reflex.Backend.Socket.Accept+ , Reflex.Backend.Socket.Connect+ , Reflex.Backend.Socket.Error+ build-depends: base >= 4.12 && < 4.13+ , bytestring >= 0.10 && < 0.11+ , lens >= 4.15.4 && < 4.19+ , mtl >= 2.2 && < 2.3+ , network >= 2.6 && < 3.2+ , reflex >= 0.5 && < 0.7+ , semialign >= 1 && < 1.2+ , semigroupoids >= 5.2.2 && < 5.4+ , stm >= 2.4 && < 2.6+ , these >= 1 && < 1.1+ hs-source-dirs: src+ ghc-options: -Wall+ default-language: Haskell2010++executable example-server+ main-is: Server.hs+ build-depends: base >= 4.12 && < 4.13+ , bytestring >= 0.10 && < 0.11+ , containers >= 0.5 && < 0.7+ , lens >= 4.15.4 && < 4.19+ , network >= 2.6 && < 3.2+ , reflex >= 0.5 && < 0.7+ , reflex-backend-socket+ , reflex-basic-host >= 0.2 && < 0.3+ hs-source-dirs: example+ ghc-options: -Wall -threaded+ default-language: Haskell2010++executable example-client+ main-is: Client.hs+ build-depends: base >= 4.12 && < 4.13+ , bytestring >= 0.10 && < 0.11+ , network >= 2.6 && < 3.2+ , reflex >= 0.5 && < 0.7+ , reflex-backend-socket+ , reflex-basic-host >= 0.2 && < 0.3+ hs-source-dirs: example+ ghc-options: -Wall -threaded+ default-language: Haskell2010+++executable example-others+ main-is: Others.hs+ build-depends: base >= 4.12 && < 4.13+ , bytestring >= 0.10 && < 0.11+ , network >= 2.6 && < 3.2+ , reflex >= 0.5 && < 0.7+ , reflex-backend-socket+ , reflex-basic-host >= 0.2 && < 0.3+ , witherable >= 0.2 && < 0.4+ hs-source-dirs: example+ ghc-options: -Wall -threaded+ default-language: Haskell2010
+ src/Reflex/Backend/Socket.hs view
@@ -0,0 +1,210 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeApplications #-}++{-|+Module : Reflex.Backend.Socket+Description : Wrap a TCP socket to communicate through @Event t ByteString@+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable++Use 'socket' to wrap a network 'Socket' so that it sends out the+firings of an @'Event' t 'ByteString'@, and fires any data that it+receives on another @'Event' t 'ByteString'@.+-}++module Reflex.Backend.Socket+ ( socket++ -- * Socket configuration+ , SocketConfig(..)++ -- * Socket output events+ , Socket(..)++ -- * Lenses+ -- ** 'SocketConfig'+ , scInitSocket+ , scMaxRx+ , scSend+ , scClose++ -- ** 'Socket'+ , sReceive+ , sOpen+ , sClose+ , sError++ -- * Convenience re-exports+ , module Reflex.Backend.Socket.Accept+ , module Reflex.Backend.Socket.Connect+ , module Reflex.Backend.Socket.Error+ ) where++import Control.Concurrent (forkIO)+import qualified Control.Concurrent.STM as STM+import Control.Exception (IOException, try)+import Control.Lens.TH (makeLenses)+import Control.Monad.IO.Class (MonadIO(..))+import Control.Monad.STM (atomically)+import Data.Align (align)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.Functor (($>), (<&>), void)+import Data.These+import qualified Network.Socket as NS+import Network.Socket.ByteString (sendAll, recv)+import Reflex+import Reflex.Backend.Socket.Accept+import Reflex.Backend.Socket.Connect+import Reflex.Backend.Socket.Error++-- | Holds the socket to wire into the FRP network, and events that+-- drive it.+data SocketConfig t = SocketConfig+ { _scInitSocket :: NS.Socket+ -- ^ Socket to wrap.+ , _scMaxRx :: Int+ -- ^ Maximum number of bytes to read at a time.+ , _scSend :: Event t ByteString+ -- ^ Data to send out on this socket.+ , _scClose :: Event t ()+ -- ^ Ask to close the socket. The socket will stop trying to+ -- receive data (and the '_sReceive' event will stop firing), and+ -- the socket will be "drained": future events on '_scSend' will+ -- be ignored, and it will close after writing all pending data.+ -- If '_scSend' and '_scClose' fire in the same frame, the data+ -- will nevertheless be queued for sending.+ }++$(makeLenses ''SocketConfig)++-- | Events produced by an active socket.+data Socket t = Socket+ { _sReceive :: Event t ByteString+ -- ^ Data has arrived.+ , _sOpen :: Event t ()+ -- ^ The socket has opened, and its receive/send loops are running.+ , _sClose :: Event t ()+ -- ^ The socket has closed. This will fire exactly once when the+ -- socket closes for any reason, including if your '_scClose'+ -- event fires, the other end disconnects, or if the socket closes+ -- in response to a caught exception.+ , _sError :: Event t IOException+ -- ^ An exception occurred. Treat the socket as closed after you+ -- see this. If the socket was open, you will see the '_sClose'+ -- event fire as well, but not necessarily in the same frame.+ }++$(makeLenses ''Socket)++data SocketState+ = Open+ -- ^ Data flows in both directions+ | Draining+ -- ^ We've been asked to close, but will transmit all pending data+ -- first (and not accept any more)+ | Closed+ -- ^ Hard close. Don't transmit pending data.++-- | Wire a socket into the FRP network. You will likely use this to+-- attach events to a socket that you just connected (from+-- 'Reflex.Backend.Socket.Connect.connect'), or a socket that you just+-- accepted (from the 'Reflex.Backend.Socket.Accept._aAcceptSocket'+-- event you got when you called+-- 'Reflex.Backend.Socket.Accept.accept').+socket+ :: forall t m.+ ( Reflex t+ , PerformEvent t m+ , PostBuild t m+ , TriggerEvent t m+ , MonadIO (Performable m)+ , MonadIO m+ )+ => SocketConfig t+ -> m (Socket t)+socket (SocketConfig sock maxRx eTx eClose) = do+ (eRx, onRx) <- newTriggerEvent+ (eOpen, onOpen) <- newTriggerEvent+ (eClosed, onClosed) <- newTriggerEvent+ (eError, onError) <- newTriggerEvent++ payloadQueue <- liftIO STM.newTQueueIO+ state <- liftIO $ STM.newTVarIO Open++ let+ start = liftIO $ do+ void $ forkIO txLoop+ void $ forkIO rxLoop+ void $ forkIO closeSentinel+ onOpen ()++ where+ txLoop =+ let+ loop = do+ mBytes <- atomically $+ STM.readTVar state >>= \case+ Closed -> pure Nothing+ Draining -> STM.tryReadTQueue payloadQueue+ Open -> STM.tryReadTQueue payloadQueue+ >>= maybe STM.retry (pure . Just)++ case mBytes of+ Nothing -> shutdown+ Just bs ->+ try (sendAll sock bs) >>= \case+ Left exc -> onError exc *> shutdown+ Right () -> loop+ in loop++ rxLoop =+ let+ loop = atomically (STM.readTVar state) >>= \case+ Open -> try (recv sock maxRx) >>= \case+ Left exc -> onError exc *> shutdown+ Right bs+ | B.null bs -> shutdown+ | otherwise -> onRx bs *> loop+ _ -> pure ()+ in loop++ closeSentinel = do+ atomically $ STM.readTVar state >>= \case+ Closed -> pure ()+ _ -> STM.retry++ void . try @IOException $ NS.close sock+ onClosed ()++ shutdown = void . atomically $ STM.writeTVar state Closed++ ePostBuild <- getPostBuild+ performEvent_ $ ePostBuild $> start++ -- If we see a tx and a close event in the same frame, we want to+ -- process the tx before the close, so it doesn't get lost.+ let+ eTxOrClose :: Event t (These ByteString ())+ eTxOrClose = align eTx eClose++ queueSend bs = STM.readTVar state >>= \case+ Open -> STM.writeTQueue payloadQueue bs+ _ -> pure ()++ queueClose = STM.modifyTVar state $ \case+ Open -> Draining+ s -> s++ performEvent_ $ eTxOrClose <&> liftIO . atomically . \case+ This bs -> queueSend bs+ That () -> queueClose+ These bs () -> queueSend bs *> queueClose++ pure $ Socket eRx eOpen eClosed eError
+ src/Reflex/Backend/Socket/Accept.hs view
@@ -0,0 +1,174 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-}++{-|+Module : Reflex.Backend.Socket.Accept+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable++Use 'accept' to create listen sockets, and get an @'Event' t 'Socket'@+of new connections.+-}++module Reflex.Backend.Socket.Accept+ ( accept++ -- * Listen socket configuration+ , AcceptConfig(..)++ -- * Results of @accept@+ , Accept(..)++ -- * Lenses+ -- ** 'AcceptConfig'+ , acHostname+ , acService+ , acListenQueue+ , acSocketOptions+ , acClose++ -- ** 'Accept'+ , aAcceptSocket+ , aClose+ , aError+ ) where++import Control.Concurrent (forkIO)+import qualified Control.Concurrent.STM as STM+import Control.Exception (IOException, onException, try)+import Control.Lens.TH (makeLenses)+import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT)+import Control.Monad.STM (atomically)+import Control.Monad.Trans (MonadIO(..))+import Data.Foldable (traverse_)+import Data.Functor (($>), void)+import Data.List.NonEmpty (NonEmpty, fromList)+import Data.Semigroup.Foldable (asum1)+import Network.Socket (AddrInfo(..), AddrInfoFlag(..), Socket)+import qualified Network.Socket as NS+import Reflex+import Reflex.Backend.Socket.Error (SetupError(..))++-- | Configuration of a listen socket.+data AcceptConfig t = AcceptConfig+ { _acHostname :: Maybe NS.HostName+ -- ^ The hostname to bind to. This will often be 'Nothing', to+ -- bind all interfaces.+ , _acService :: Maybe NS.ServiceName+ -- ^ The port number or service name to listen on. See the+ -- <https://linux.die.net/man/3/getaddrinfo manpage for getaddrinfo>.+ , _acListenQueue :: Int+ -- ^ The length of the "pending connections" queue. See the+ -- <https://linux.die.net/man/2/listen manpage for listen>.+ , _acSocketOptions :: [(NS.SocketOption, Int)]+ -- ^ List of socket options, passed one at a time to+ -- 'NS.setSocketOption'. Many people will want+ -- @[('NS.ReuseAddr', 1)]@ here.+ , _acClose :: Event t ()+ -- ^ Close the listen socket when this event fires. If you plan to+ -- run forever, use 'never'.+ }++$(makeLenses ''AcceptConfig)++-- | Events produced by a running listen socket.+data Accept t = Accept+ { _aAcceptSocket :: Event t (Socket, NS.SockAddr)+ -- ^ A new connection was accepted, including its remote address.+ , _aClose :: Event t ()+ -- ^ The socket has closed. This will fire exactly once when the+ -- socket closes for any reason, including if your '_acClose'+ -- event fires or if the socket closes in response to a caught+ -- exception.+ , _aError :: Event t IOException+ -- ^ An exception occurred. Treat the socket as closed after you+ -- see this. You will see the '_aClose' event fire as well, but+ -- not necessarily in the same frame.+ }++$(makeLenses ''Accept)++-- | Create a listening socket. Sockets are accepted in a background+-- thread.+accept+ :: ( Reflex t+ , PerformEvent t m+ , PostBuild t m+ , TriggerEvent t m+ , MonadIO (Performable m)+ , MonadIO m+ )+ => AcceptConfig t+ -> m (Event t (Either SetupError (Accept t)))+ -- ^ This event will fire exactly once.+accept (AcceptConfig mHost mService listenQueue options eClose) = do+ (eAccept, onAccept) <- newTriggerEvent+ (eClosed, onClosed) <- newTriggerEvent+ (eError, onError) <- newTriggerEvent++ isOpen <- liftIO STM.newEmptyTMVarIO++ let+ start = liftIO $ makeListenSocket >>= \case+ Left exc -> pure $ Left exc+ Right sock -> do+ atomically $ STM.putTMVar isOpen sock+ void $ forkIO acceptLoop+ pure . Right $ Accept eAccept eClosed eError++ where+ makeListenSocket =+ let+ getAddrs :: ExceptT SetupError IO (NonEmpty AddrInfo)+ getAddrs = withExceptT GetAddrInfoError . ExceptT . try $+ -- fromList is OK here, as getaddrinfo(3) is required to+ -- return a nonempty list of addrinfos.+ --+ -- See: http://pubs.opengroup.org/onlinepubs/9699919799/functions/getaddrinfo.html+ -- And: https://github.com/haskell/network/issues/407+ fromList <$> NS.getAddrInfo (Just passiveHints) mHost mService+ where passiveHints = NS.defaultHints { addrFlags = [AI_PASSIVE] }++ tryListen+ :: AddrInfo+ -> ExceptT (NonEmpty (AddrInfo, IOException)) IO Socket+ tryListen info = withExceptT (pure . (info,)) . ExceptT . try $ do+ sock <- NS.socket (addrFamily info) NS.Stream NS.defaultProtocol+ (`onException` NS.close sock) $ do+ traverse_ (uncurry $ NS.setSocketOption sock) options+ NS.bind sock (addrAddress info)+ NS.listen sock listenQueue+ pure sock++ in runExceptT $ do+ addrs <- getAddrs+ let attempts = tryListen <$> addrs+ withExceptT UseAddrInfoError $ asum1 attempts++ acceptLoop =+ let+ -- If we receive an exception when trying to accept, check+ -- the TMVar that's meant to hold our socket. If it's+ -- empty, then 'eClose' must have fired (and the socket+ -- closed under us) and we should go quietly. Otherwise,+ -- close the socket ourselves and signal 'eError'.+ exHandlerAccept e = atomically (STM.tryReadTMVar isOpen)+ >>= maybe (pure ()) (const $ close *> onError e)+ in+ atomically (STM.tryReadTMVar isOpen) >>= \case+ Nothing -> onClosed ()+ Just sock -> do+ try (NS.accept sock) >>= either exHandlerAccept onAccept+ acceptLoop++ close = atomically (STM.tryTakeTMVar isOpen) >>= traverse_ NS.close++ performEvent_ $ eClose $> liftIO close++ ePostBuild <- getPostBuild+ performEvent $ ePostBuild $> start
+ src/Reflex/Backend/Socket/Connect.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TupleSections #-}++{-|+Module : Reflex.Backend.Socket.Connect+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable++Use 'connect' to attempt a connection to a remote endpoint, and get an+@'Event' t ('Either' 'SetupError' 'Socket')@ that tells you whether or+not it worked.+-}++module Reflex.Backend.Socket.Connect (connect) where++import Control.Concurrent (forkIO)+import Control.Exception (IOException, onException, try)+import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT)+import Control.Monad.Trans (MonadIO(..))+import Data.Functor (($>), void)+import Data.List.NonEmpty (NonEmpty, fromList)+import Data.Semigroup.Foldable (asum1)+import Network.Socket (Socket, AddrInfo(..), defaultProtocol)+import qualified Network.Socket as NS+import Reflex+import Reflex.Backend.Socket.Error (SetupError(..))++-- | Connect to a remote endpoint. The connection happens in a+-- background thread.+connect+ :: ( Reflex t+ , PerformEvent t m+ , TriggerEvent t m+ , PostBuild t m+ , MonadIO (Performable m)+ , MonadIO m+ )+ => Maybe NS.HostName+ -- ^ Host to connect to. If 'Nothing', connect via loopback.+ -> NS.ServiceName+ -- ^ Service (port number or service name). See the+ -- <https://linux.die.net/man/3/getaddrinfo manpage for getaddrinfo>.+ -> m (Event t (Either SetupError Socket))+ -- ^ This event will fire exactly once.+connect mHost service = do+ ePostBuild <- getPostBuild+ performEventAsync $ ePostBuild $> \onRes -> void . liftIO . forkIO $+ let+ getAddrs :: ExceptT SetupError IO (NonEmpty AddrInfo)+ getAddrs = withExceptT GetAddrInfoError . ExceptT . try $+ -- fromList is OK here, as getaddrinfo(3) is required to+ -- return a nonempty list of addrinfos.+ --+ -- See: http://pubs.opengroup.org/onlinepubs/9699919799/functions/getaddrinfo.html+ -- And: https://github.com/haskell/network/issues/407+ fromList <$> NS.getAddrInfo Nothing mHost (Just service)++ tryConnect+ :: AddrInfo+ -> ExceptT (NonEmpty (AddrInfo, IOException)) IO Socket+ tryConnect info = withExceptT (pure . (info,)) . ExceptT . try $ do+ sock <- NS.socket (addrFamily info) NS.Stream defaultProtocol+ NS.connect sock (addrAddress info) `onException` NS.close sock+ pure sock++ in do+ res <- runExceptT $ do+ addrs <- getAddrs+ let attempts = tryConnect <$> addrs+ withExceptT UseAddrInfoError $ asum1 attempts+ onRes res
+ src/Reflex/Backend/Socket/Error.hs view
@@ -0,0 +1,45 @@+{-|+Module : Reflex.Backend.Socket.Error+Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation+License : BSD3+Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au+Stability : experimental+Portability : non-portable+-}++{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE TemplateHaskell #-}++module Reflex.Backend.Socket.Error+ ( SetupError(..)++ -- * Prisms+ -- ** 'SetupError'+ , _GetAddrInfoError+ , _UseAddrInfoError+ ) where++import Control.Exception (IOException)+import Data.List.NonEmpty (NonEmpty)+import GHC.Generics (Generic)+import Network.Socket (AddrInfo)+import Control.Lens.TH (makePrisms)++-- | If a "socket setup" fails ('Reflex.Backend.Socket.Accept.accept'+-- or 'Reflex.Backend.Socket.Connect.connect'), you'll inspect one of+-- these to find out why.+data SetupError+ = GetAddrInfoError IOException+ -- ^ Call to 'Network.Socket.getAddrInfo' failed.+ | UseAddrInfoError (NonEmpty (AddrInfo, IOException))+ -- ^ We failed to set up a socket with any 'AddrInfo' we were+ -- given, and here are the corresponding exceptions each time we+ -- tried. For 'Reflex.Backend.Socket.Accept.accept', this means+ -- either 'Network.Socket.bind' or 'Network.Socket.listen' failed,+ -- and for 'Reflex.Backend.Socket.Connect.connect', this means+ -- 'Network.Socket.Connect' failed.+ deriving (Eq, Generic, Show)++$(makePrisms ''SetupError)