git.iko.soy/streamly
on 8ecf96a help sign in
streamly/src/Streamly/Network/Client.hs
152 lines 4.6 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE UnboxedTuples #-}
8#include "inline.hs"
10-- |
11-- Module : Streamly.Network.Client
12-- Copyright : (c) 2019 Composewell Technologies
13--
14-- License : BSD3
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
17-- Portability : GHC
18--
19-- Combinators to build network clients.
20--
21-- > import qualified Streamly.Network.Socket as Client
22--
24module Streamly.Network.Client
25 (
26 -- * Interact
27 -- | Socket based reads and writes.
28 withConnection
30 -- * Source
31 , read
32 -- , readUtf8
33 -- , readLines
34 -- , readFrames
35 -- , readByChunks
37 -- -- * Array Read
38 -- , readArrayUpto
39 -- , readArrayOf
41 -- , readArraysUpto
42 -- , readArraysOf
43 -- , readArrays
45 -- * Sink
46 , write
47 -- , writeUtf8
48 -- , writeUtf8ByLines
49 -- , writeByFrames
50 -- , writeByChunks
52 -- -- * Array Write
53 -- , writeArray
54 , writeArrays
55 )
56where
58import Control.Monad.Catch (MonadCatch)
59import Control.Monad.IO.Class (MonadIO(..))
60import Data.Word (Word8)
61import Network.Socket
62 (Socket, PortNumber, Family(..), SockAddr(..), SocketType(..),
63 socket, connect, defaultProtocol)
64import Prelude hiding (read)
66import qualified Network.Socket as Net
68import Streamly (MonadAsync)
69import Streamly.Mem.Array.Types (Array(..), defaultChunkSize)
70import Streamly.Streams.Serial (SerialT)
71import Streamly.Streams.StreamK.Type (IsStream)
73import qualified Streamly.Mem.Array as A
74import qualified Streamly.Prelude as S
75import qualified Streamly.Network.Socket as SK
77-------------------------------------------------------------------------------
78-- Connect
79-------------------------------------------------------------------------------
81openConnection :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
82openConnection addr port = do
83 sock <- socket AF_INET Stream defaultProtocol
84 connect sock $ SockAddrInet port (Net.tupleToHostAddress addr)
85 return sock
87-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
88-- host address and port and passes the resulting socket handle to the
89-- computation @act@. The handle will be closed on exit from 'withConnection',
90-- whether by normal termination or by raising an exception. If closing the
91-- handle raises an exception, then this exception will be raised by
92-- 'withConnection' rather than any exception raised by 'act'.
93--
94-- @since 0.7.0
95{-# INLINABLE withConnection #-}
96withConnection :: (IsStream t, MonadCatch m, MonadIO m)
97 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
98withConnection addr port =
99 S.bracket (liftIO $ openConnection addr port) (liftIO . Net.close)
101-------------------------------------------------------------------------------
102-- Read Addr to Stream
103-------------------------------------------------------------------------------
105-- | Read a stream from the supplied IPv4 host address and port number.
106--
107-- @since 0.7.0
108{-# INLINE read #-}
109read :: (IsStream t, MonadCatch m, MonadIO m)
110 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
111read addr port = A.flattenArrays $ withConnection addr port SK.readArrays
113-------------------------------------------------------------------------------
114-- Writing
115-------------------------------------------------------------------------------
117-- | Write a stream of arrays to the supplied IPv4 host address and port
118-- number.
119--
120-- @since 0.7.0
121{-# INLINE writeArrays #-}
122writeArrays
123 :: (MonadCatch m, MonadAsync m)
124 => (Word8, Word8, Word8, Word8)
125 -> PortNumber
126 -> SerialT m (Array Word8)
127 -> m ()
128writeArrays addr port xs =
129 S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)
131-- | Like 'write' but provides control over the write buffer. Output will
132-- be written to the IO device as soon as we collect the specified number of
133-- input elements.
134--
135-- @since 0.7.0
136{-# INLINE writeByChunks #-}
137writeByChunks
138 :: (MonadCatch m, MonadAsync m)
139 => Int
140 -> (Word8, Word8, Word8, Word8)
141 -> PortNumber
142 -> SerialT m Word8
143 -> m ()
144writeByChunks n addr port m = writeArrays addr port $ A.arraysOf n m
146-- | Write a stream to the supplied IPv4 host address and port number.
147--
148-- @since 0.7.0
149{-# INLINE write #-}
150write :: (MonadCatch m, MonadAsync m)
151 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
152write = writeByChunks defaultChunkSize