@@ -0,0 +1,202 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
7+
8+#include "inline.hs"
9+
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+--
23+
24+module Streamly.Internal.Network.Client
25+ (
26+ -- * Interact
27+ -- | Socket based reads and writes.
28+ withConnection
29+
30+ -- * Source
31+ , read
32+ -- , readUtf8
33+ -- , readLines
34+ -- , readFrames
35+ -- , readByChunks
36+
37+ -- -- * Array Read
38+ -- , readArrayUpto
39+ -- , readArrayOf
40+
41+ -- , readArraysUpto
42+ -- , readArraysOf
43+ -- , readArrays
44+
45+ -- * Sink
46+ , write
47+ -- , writeUtf8
48+ -- , writeUtf8ByLines
49+ -- , writeByFrames
50+ -- , writeInChunksOf
51+
52+ -- -- * Array Write
53+ -- , writeArray
54+ , writeArrays
55+ )
56+where
57+
58+import Control.Monad.Catch (MonadCatch)
59+import Control.Monad.IO.Class (MonadIO(..))
60+import Data.Word (Word8)
61+import Network.Socket
62+ (Socket, PortNumber, Family(..), SockAddr(..), SocketType(..),
63+ socket, connect, defaultProtocol)
64+import Prelude hiding (read)
65+
66+import qualified Network.Socket as Net
67+
68+import Streamly (MonadAsync)
69+import Streamly.Internal.Data.Fold.Types (Fold(..))
70+import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
71+-- import Streamly.Streams.Serial (SerialT)
72+import Streamly.Streams.StreamK.Type (IsStream)
73+
74+import qualified Streamly.Internal.Memory.ArrayStream as AS
75+import qualified Streamly.Internal.Data.Fold.Types as FL
76+import qualified Streamly.Prelude as S
77+import qualified Streamly.Network.Socket as SK
78+import qualified Streamly.Internal.Network.Socket as ISK
79+
80+-------------------------------------------------------------------------------
81+-- Connect
82+-------------------------------------------------------------------------------
83+
84+openConnection :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
85+openConnection addr port = do
86+ sock <- socket AF_INET Stream defaultProtocol
87+ connect sock $ SockAddrInet port (Net.tupleToHostAddress addr)
88+ return sock
89+
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 #-}
99+withConnection :: (IsStream t, MonadCatch m, MonadIO m)
100+ => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
101+withConnection addr port =
102+ S.bracket (liftIO $ openConnection addr port) (liftIO . Net.close)
103+
104+-------------------------------------------------------------------------------
105+-- Read Addr to Stream
106+-------------------------------------------------------------------------------
107+
108+-- | Read a stream from the supplied IPv4 host address and port number.
109+--
110+-- @since 0.7.0
111+{-# INLINE read #-}
112+read :: (IsStream t, MonadCatch m, MonadIO m)
113+ => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
114+read addr port = AS.concat $ withConnection addr port ISK.toStreamArrays
115+
116+-------------------------------------------------------------------------------
117+-- Writing
118+-------------------------------------------------------------------------------
119+
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 #-}
126+writeArrays
127+ :: (MonadCatch m, MonadAsync m)
128+ => (Word8, Word8, Word8, Word8)
129+ -> PortNumber
130+ -> SerialT m (Array Word8)
131+ -> m ()
132+writeArrays addr port xs =
133+ S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)
134+-}
135+
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 #-}
141+writeArrays
142+ :: (MonadAsync m)
143+ => (Word8, Word8, Word8, Word8)
144+ -> PortNumber
145+ -> Fold m (Array Word8) ()
146+writeArrays 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
153+
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 #-}
161+writeInChunksOf
162+ :: (MonadCatch m, MonadAsync m)
163+ => Int
164+ -> (Word8, Word8, Word8, Word8)
165+ -> PortNumber
166+ -> SerialT m Word8
167+ -> m ()
168+writeInChunksOf n addr port m = writeArrays addr port $ AS.arraysOf n m
169+-}
170+
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 #-}
177+writeInChunksOf
178+ :: (MonadAsync m)
179+ => Int
180+ -> (Word8, Word8, Word8, Word8)
181+ -> PortNumber
182+ -> Fold m Word8 ()
183+writeInChunksOf n addr port =
184+ FL.lchunksOf n (writeNUnsafe n) (writeArrays addr port)
185+
186+{-
187+-- | Write a stream to the supplied IPv4 host address and port number.
188+--
189+-- @since 0.7.0
190+{-# INLINE write #-}
191+write :: (MonadCatch m, MonadAsync m)
192+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
193+write = writeInChunksOf defaultChunkSize
194+-}
195+
196+-- | Write a stream to the supplied IPv4 host address and port number.
197+--
198+-- @since 0.7.0
199+{-# INLINE write #-}
200+write :: MonadAsync m
201+ => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
202+write = writeInChunksOf defaultChunkSize