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