git.iko.soy/streamly
on d098a05 help sign in
streamly/src/Streamly/Network/Client.hs
201 lines 5.9 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 -- , writeInChunksOf
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.Internal.Data.Fold.Types (Fold(..))
70import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
71-- import Streamly.Streams.Serial (SerialT)
72import Streamly.Streams.StreamK.Type (IsStream)
74import qualified Streamly.Internal.Memory.ArrayStream as AS
75import qualified Streamly.Internal.Data.Fold.Types as FL
76import qualified Streamly.Prelude as S
77import qualified Streamly.Network.Socket as SK
79-------------------------------------------------------------------------------
80-- Connect
81-------------------------------------------------------------------------------
83openConnection :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
84openConnection addr port = do
85 sock <- socket AF_INET Stream defaultProtocol
86 connect sock $ SockAddrInet port (Net.tupleToHostAddress addr)
87 return sock
89-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
90-- host address and port and passes the resulting socket handle to the
91-- computation @act@. The handle will be closed on exit from 'withConnection',
92-- whether by normal termination or by raising an exception. If closing the
93-- handle raises an exception, then this exception will be raised by
94-- 'withConnection' rather than any exception raised by 'act'.
95--
96-- @since 0.7.0
97{-# INLINABLE withConnection #-}
98withConnection :: (IsStream t, MonadCatch m, MonadIO m)
99 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
100withConnection addr port =
101 S.bracket (liftIO $ openConnection addr port) (liftIO . Net.close)
103-------------------------------------------------------------------------------
104-- Read Addr to Stream
105-------------------------------------------------------------------------------
107-- | Read a stream from the supplied IPv4 host address and port number.
108--
109-- @since 0.7.0
110{-# INLINE read #-}
111read :: (IsStream t, MonadCatch m, MonadIO m)
112 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
113read addr port = AS.concat $ withConnection addr port SK.readArrays
115-------------------------------------------------------------------------------
116-- Writing
117-------------------------------------------------------------------------------
119{-
120-- | Write a stream of arrays to the supplied IPv4 host address and port
121-- number.
122--
123-- @since 0.7.0
124{-# INLINE writeArrays #-}
125writeArrays
126 :: (MonadCatch m, MonadAsync m)
127 => (Word8, Word8, Word8, Word8)
128 -> PortNumber
129 -> SerialT m (Array Word8)
130 -> m ()
131writeArrays addr port xs =
132 S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)
133-}
135-- | Write a stream of arrays to the supplied IPv4 host address and port
136-- number.
137--
138-- @since 0.7.0
139{-# INLINE writeArrays #-}
140writeArrays
141 :: (MonadAsync m)
142 => (Word8, Word8, Word8, Word8)
143 -> PortNumber
144 -> Fold m (Array Word8) ()
145writeArrays addr port = Fold step initial extract
146 where
147 initial = do
148 skt <- liftIO (openConnection addr port)
149 FL.initialize (SK.writeArrays skt)
150 step = FL.runStep
151 extract (Fold _ initial1 extract1) = initial1 >>= extract1
153{-
154-- | Like 'write' but provides control over the write buffer. Output will
155-- be written to the IO device as soon as we collect the specified number of
156-- input elements.
157--
158-- @since 0.7.0
159{-# INLINE writeInChunksOf #-}
160writeInChunksOf
161 :: (MonadCatch m, MonadAsync m)
162 => Int
163 -> (Word8, Word8, Word8, Word8)
164 -> PortNumber
165 -> SerialT m Word8
166 -> m ()
167writeInChunksOf n addr port m = writeArrays addr port $ AS.arraysOf n m
168-}
170-- | Like 'write' but provides control over the write buffer. Output will
171-- be written to the IO device as soon as we collect the specified number of
172-- input elements.
173--
174-- @since 0.7.0
175{-# INLINE writeInChunksOf #-}
176writeInChunksOf
177 :: (MonadAsync m)
178 => Int
179 -> (Word8, Word8, Word8, Word8)
180 -> PortNumber
181 -> Fold m Word8 ()
182writeInChunksOf n addr port =
183 FL.lchunksOf n (writeNUnsafe n) (writeArrays addr port)
185{-
186-- | Write a stream to the supplied IPv4 host address and port number.
187--
188-- @since 0.7.0
189{-# INLINE write #-}
190write :: (MonadCatch m, MonadAsync m)
191 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
192write = writeInChunksOf defaultChunkSize
193-}
195-- | Write a stream to the supplied IPv4 host address and port number.
196--
197-- @since 0.7.0
198{-# INLINE write #-}
199write :: MonadAsync m
200 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
201write = writeInChunksOf defaultChunkSize