pipes-p2p 0.1 → 0.2

raw patch · 3 files changed

+30/−5 lines, 3 files

Files

• CHANGELOG +3/−0
• pipes-p2p.cabal +3/−1
• src/Pipes/Network/P2P.hs +24/−4

@@ -0,0 +1,3 @@+0.2+---+* Make sure specified number of bytes are retrieved from socket.

@@ -1,5 +1,5 @@ name:                pipes-p2p-version:             0.1+version:             0.2 cabal-version:       >=1.10 tested-with:         GHC == 7.6.3 build-type:          Simple@@ -12,6 +12,8 @@ synopsis:            P2P network nodes with pipes description:   Toy library to facilitate the creation of custom P2P networks using `pipes`.++extra-source-files: CHANGELOG  source-repository head     type: git

@@ -1,9 +1,9 @@-{-# OPTIONS_HADDOCK ignore-exports #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE BangPatterns #-}  {-| Use 'node' to create a 'Node' with your desired settings and then 'launch'     it.@@ -36,7 +36,7 @@ import Control.Applicative (Applicative, (<$>)) import Control.Monad (forever, void, unless, guard) import Data.Monoid ((<>))-import Data.Foldable (for_, forM_)+import Data.Foldable (for_) import Control.Concurrent (ThreadId, myThreadId) import Control.Concurrent.Async (async, link) import GHC.Generics (Generic)@@ -277,6 +277,7 @@  instance Binary Header +-- | Byte length of 'Header'. hSize :: Int hSize = B.length . encode $ Header 0 0 {-# INLINE hSize #-}@@ -284,15 +285,23 @@ -- ** Socket reader  socketReader :: (MonadIO m, Binary a) => Int -> Socket -> Producer a m ()-socketReader magic sock = fromSocketN sock >+> beheader magic >+> decoder $ ()+socketReader magic sock = fromSocketN sock >+> exhaust >+> beheader magic >+> decoder $ () {-# INLINABLE socketReader #-} +{- | Decodes 'ByteString's flowing downstream.++     In case of decoding failure, it discards the bytes and tries to decode+     again from upstream.+-} decoder :: (MonadIO m, Binary a) => () -> Pipe ByteString a m () decoder () = forever $ do     pbs <- await-    forM_ (decode pbs) yield+    for_ (decode pbs) yield {-# INLINABLE decoder #-} +{-| Pulls header bytes, checks for magic bytes and pulls the payload based on+    the expected length in the 'Header'.+-} beheader :: MonadIO m => Int -> () -> Proxy Int ByteString () ByteString m () beheader magic () = forever $ do     hbs <- request hSize@@ -301,6 +310,17 @@         Just (Header magic' nbytes) -> unless (magic /= magic')                                      $ request nbytes >>= respond {-# INLINABLE beheader #-}++-- | Make sure the number of specified bytes are received from upstream.+exhaust :: MonadIO m => Int -> Proxy Int ByteString Int ByteString m ()+exhaust n0 = forever $ go B.empty n0+  where+    go !acc n = do bs <- request n+                   let rl = B.length bs+                   if rl == n+                   then respond $ acc <> bs+                   else go (acc <> bs) (n - rl)+{-# INLINABLE exhaust #-}  -- ** Strict Binary