potoki-core 0.12 → 1.2

raw patch · 10 files changed

+294/−529 lines, 10 filesdep +dequedep −acquirePVP ok

version bump matches the API change (PVP)

Dependencies added: deque

Dependencies removed: acquire

API changes (from Hackage documentation)

Files

• LICENSE +1/−1
• library/Potoki/Core/Consume.hs +55/−76
• library/Potoki/Core/Fetch.hs +58/−153
• library/Potoki/Core/IO.hs +27/−41
• library/Potoki/Core/Prelude.hs +1/−7
• library/Potoki/Core/Produce.hs +25/−47
• library/Potoki/Core/Transform.hs +98/−104
• library/Potoki/Core/Types.hs +0/−34
• potoki-core.cabal +14/−12
• tests/Main.hs +15/−54

@@ -1,4 +1,4 @@-Copyright (c) 2017, Metrix.AI+Copyright (c) 2017, Nikita Volkov  Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation

@@ -1,98 +1,77 @@-module Potoki.Core.Consume-(-  Consume(..),-  apConcurrently,-  list,-  sum,-  transform,-)-where+module Potoki.Core.Consume where -import Potoki.Core.Prelude hiding (sum)-import Potoki.Core.Types+import Potoki.Core.Prelude import qualified Potoki.Core.Fetch as A-import qualified Acquire.IO as B  +{-|+Active consumer of input into output.+Sort of like a reducer in Map/Reduce.++Automates the management of resources.+-}+newtype Consume input output =+  {-|+  An action, which executes the provided fetch in IO,+  while managing the resources behind the scenes.+  -}+  Consume (A.Fetch input -> IO output)+ instance Profunctor Consume where   {-# INLINE dimap #-}   dimap inputMapping outputMapping (Consume consume) =-    Consume (\ fetch -> fmap outputMapping (consume $ fmap inputMapping fetch))+    Consume (\ fetch -> fmap outputMapping (consume (fmap inputMapping fetch)))  instance Choice Consume where-  right' :: Consume a b -> Consume (Either c a) (Either c b)   right' (Consume rightConsumeIO) =-     Consume $ \ (Fetch eitherFetchIO) -> do-       fetchedLeftMaybeRef <- newIORef Nothing-       consumedRight <--         rightConsumeIO $ Fetch $ do-           eitherFetch <- eitherFetchIO-           case eitherFetch of-             Nothing      -> return Nothing-             Just element -> case element of-               Right fetchedRight -> return $ Just fetchedRight-               Left  fetchedLeft  -> do-                 writeIORef fetchedLeftMaybeRef $ Just fetchedLeft-                 return Nothing-       fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef-       case fetchedLeftMaybe of-         Nothing          -> return $ Right consumedRight-         Just fetchedLeft -> return $ Left fetchedLeft +    Consume $ \ (A.Fetch eitherFetchIO) -> do+      fetchedLeftMaybeRef <- newIORef Nothing+      consumedRight <- +        rightConsumeIO $ A.Fetch $ \ nil just -> join $ eitherFetchIO (return nil) $ \ case+          Right !fetchedRight -> return (just fetchedRight)+          Left !fetchedLeft -> writeIORef fetchedLeftMaybeRef (Just fetchedLeft) >> return nil+      fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef+      case fetchedLeftMaybe of+        Nothing -> return (Right consumedRight)+        Just fetchedLeft -> return (Left fetchedLeft)  instance Functor (Consume input) where   fmap = rmap -instance Applicative (Consume a) where-  pure x = Consume $ \ _ -> pure x--  Consume leftConsumeIO <*> Consume rightConsumeIO =-    Consume $ \ fetch -> leftConsumeIO fetch <*> rightConsumeIO fetch--instance Monad (Consume a) where-  Consume leftConsumeIO >>= toRightConsumeIO = Consume $ \ fetch -> do-    Consume rightConsumeIO <- toRightConsumeIO <$> leftConsumeIO fetch-    rightConsumeIO fetch--instance MonadIO (Consume a) where-  liftIO a = Consume $ \ _ -> a--apConcurrently :: Consume a (b -> c) -> Consume a b -> Consume a c-apConcurrently (Consume leftConsumeIO) (Consume rightConsumeIO) =-  Consume $ \ fetch -> do-    (leftFetch, rightFetch) <- A.duplicate fetch-    rightOutputVar <- newEmptyMVar-    _ <- forkIO $ do-      !rightOutput <- rightConsumeIO rightFetch-      putMVar rightOutputVar rightOutput-    !leftOutput <- leftConsumeIO leftFetch-    rightOutput <- takeMVar rightOutputVar-    return (leftOutput rightOutput)+instance Applicative (Consume input) where+  pure x =+    Consume (const (pure x))+  (<*>) (Consume leftConsumeIO) (Consume rightConsumeIO) =+    Consume $ \ fetch -> do+      (leftFetch, rightFetch) <- A.duplicate fetch+      rightOutputVar <- newEmptyMVar+      forkIO $ do+        !rightOutput <- rightConsumeIO rightFetch+        putMVar rightOutputVar rightOutput+      !leftOutput <- leftConsumeIO leftFetch+      rightOutput <- takeMVar rightOutputVar+      return (leftOutput rightOutput)  {-# INLINABLE list #-} list :: Consume input [input] list =-  Consume $ \ (Fetch fetchIO) ->-    let -      build !acc = do-        fetch <- fetchIO-        case fetch of-          Nothing       -> pure $ acc []-          Just !element -> build $ acc . (:) element-     in build id+  Consume $ \ (A.Fetch fetchIO) ->+  let+    build !acc =+      join+        (fetchIO+          (pure (acc []))+          (\ !element -> build (acc . (:) element)))+    in build id  {-# INLINE sum #-} sum :: Num num => Consume num num sum =-  Consume $ \ (Fetch fetchIO) ->-    let-      build !acc = do-        fetch <- fetchIO-        case fetch of-          Nothing       -> pure acc-          Just !element -> build $ element + acc-     in build 0--{-# INLINABLE transform #-}-transform :: Transform input1 input2 -> Consume input2 output -> Consume input1 output-transform (Transform transformAcquire) (Consume consumeIO) =-  Consume $ \ fetch -> B.acquire (transformAcquire fetch) consumeIO+  Consume $ \ (A.Fetch fetchIO) ->+  let+    build !acc =+      join+        (fetchIO+          (pure acc)+          (\ !element -> build (element + acc)))+    in build 0

@@ -1,47 +1,41 @@-module Potoki.Core.Fetch-(-  Fetch(..),-  duplicate,-  maybeRef,-  list,-  firstCachingSecond,-  bothFetchingFirst,-  rightHandlingLeft,-  rightCachingLeft,-  eitherFetchingRight,-  signaling,-  ioFetch,-)-where+module Potoki.Core.Fetch where  import Potoki.Core.Prelude-import Potoki.Core.Types  +{-|+Passive producer of elements.+-}+newtype Fetch element =+  {-|+  Something close to a Church encoding of @IO (Maybe element)@.+  -}+  Fetch (forall x. x -> (element -> x) -> IO x)+ deriving instance Functor Fetch  instance Applicative Fetch where   pure x =-    Fetch (pure (Just x))-  (<*>) (Fetch leftIO) (Fetch rightIO) =-    Fetch ((<*>) <$> leftIO <*> rightIO)+    Fetch (\ nil just -> pure (just x))+  (<*>) (Fetch leftFn) (Fetch rightFn) =+    Fetch (\ nil just ->+      join (leftFn (pure nil) (\ leftElement ->+        rightFn nil (\ rightElement -> just (leftElement rightElement)))))  instance Monad Fetch where   return =     pure-  (>>=) (Fetch leftIO) rightFetch =-    Fetch $ do-      leftFetching <- leftIO-      case leftFetching of-        Nothing -> return Nothing-        Just leftElement -> case rightFetch leftElement of-          Fetch rightIO -> rightIO+  (>>=) (Fetch leftFn) rightK =+    Fetch (\ nil just ->+      join (leftFn (pure nil) (\ leftElement ->+        case rightK leftElement of+          Fetch rightFn -> rightFn nil just)))  instance Alternative Fetch where   empty =-    Fetch (pure Nothing)-  (<|>) (Fetch leftIO) (Fetch rightIO) =-    Fetch ((<|>) <$> leftIO <*> rightIO)+    Fetch (\ nil just -> pure nil)+  (<|>) (Fetch leftSignal) (Fetch rightSignal) =+    Fetch (\ nil just -> join (leftSignal (rightSignal nil just) (pure . just)))  instance MonadPlus Fetch where   mzero =@@ -49,10 +43,6 @@   mplus =     (<|>) -instance MonadIO Fetch where-  liftIO io =-    Fetch (fmap Just io)- {-# INLINABLE duplicate #-} duplicate :: Fetch element -> IO (Fetch element, Fetch element) duplicate (Fetch fetchIO) =@@ -63,131 +53,46 @@     notEndVar <- newTVarIO True     let       newFetch ownBuffer mirrorBuffer =-        Fetch $ do-          fetch <- fetchIO-          join $-            atomically-             (mplus-                (do-                   element <- readTQueue ownBuffer-                   return $ return (Just element)-                )-                (do-                   notEnd <- readTVar notEndVar-                   if notEnd-                     then do-                       notFetching <- readTVar notFetchingVar-                       guard notFetching-                       writeTVar notFetchingVar False-                       return $ case fetch of-                         Nothing      -> do-                           atomically-                             (do-                                writeTVar notEndVar False-                                writeTVar notFetchingVar True-                             )-                           return Nothing-                         Just !element -> do-                           atomically-                             (do-                                writeTQueue mirrorBuffer element-                                writeTVar notFetchingVar True-                             )-                           return $ Just element-                     else return $ return Nothing-                )-             )+        Fetch+          (\ nil just -> do+            join+              (atomically+                (mplus+                  (do+                    element <- readTQueue ownBuffer+                    return (return (just element)))+                  (do+                    notEnd <- readTVar notEndVar+                    if notEnd+                      then do+                        notFetching <- readTVar notFetchingVar+                        guard notFetching+                        writeTVar notFetchingVar False+                        return+                          (join+                            (fetchIO+                              (do+                                atomically+                                  (do+                                    writeTVar notEndVar False+                                    writeTVar notFetchingVar True)+                                return nil)+                              (\ !element -> do+                                atomically+                                  (do+                                    writeTQueue mirrorBuffer element+                                    writeTVar notFetchingVar True)+                                return (just element))))+                      else return (return nil)))))       leftFetch =         newFetch leftBuffer rightBuffer       rightFetch =         newFetch rightBuffer leftBuffer-     in return (leftFetch, rightFetch)--{-# INLINABLE maybeRef #-}-maybeRef :: IORef (Maybe a) -> Fetch a-maybeRef refElem =-  Fetch $ do-    elemVal <- readIORef refElem-    case elemVal of-      Nothing      -> return Nothing-      Just element -> do-        writeIORef refElem Nothing-        return $ Just element+      in return (leftFetch, rightFetch)  {-# INLINABLE list #-} list :: IORef [element] -> Fetch element list unsentListRef =-  Fetch $ do-    refList <- readIORef unsentListRef-    case refList of-      (!headVal) : (!tailVal) -> do-        writeIORef unsentListRef tailVal-        return $ Just headVal-      _              -> do-        writeIORef unsentListRef []-        return Nothing--{-# INLINABLE firstCachingSecond #-}-firstCachingSecond :: IORef b -> Fetch (a, b) -> Fetch a-firstCachingSecond cacheRef (Fetch bothFetchIO) =-  Fetch $ do-    bothFetch <- bothFetchIO-    case bothFetch of-      Nothing                -> return Nothing-      Just (!firstVal, !secondVal) -> do-        writeIORef cacheRef secondVal-        return $ Just firstVal--{-# INLINABLE bothFetchingFirst #-}-bothFetchingFirst :: IORef b -> Fetch a -> Fetch (a, b)-bothFetchingFirst cacheRef (Fetch firstFetchIO) =-  Fetch $ do-    firstFetch <- firstFetchIO-    case firstFetch of-      Nothing            -> return Nothing-      Just !firstFetched -> do-        secondCached <- readIORef cacheRef-        return $ Just (firstFetched, secondCached)--{-# INLINABLE rightHandlingLeft #-}-rightHandlingLeft :: (left -> IO ()) -> Fetch (Either left right) -> Fetch right-rightHandlingLeft left2IO (Fetch eitherFetchIO) =-  Fetch $ do-    eitherFetch <- eitherFetchIO-    case eitherFetch of-      Nothing    -> return Nothing-      Just input -> case input of-        Right rightInput -> return $ Just rightInput-        Left  leftInput  -> left2IO leftInput $> Nothing--{-# INLINABLE rightCachingLeft #-}-rightCachingLeft :: IORef (Maybe left) -> Fetch (Either left right) -> Fetch right-rightCachingLeft cacheRef =-  rightHandlingLeft (writeIORef cacheRef . Just)--{-# INLINABLE eitherFetchingRight #-}-eitherFetchingRight :: IORef (Maybe left) -> Fetch right -> Fetch (Either left right)-eitherFetchingRight cacheRef (Fetch rightFetchIO) =-  Fetch $ do-    rightFetch <- rightFetchIO-    case rightFetch of-      Nothing -> return Nothing-      Just r  -> atomicModifyIORef' cacheRef $ \ case-        Nothing -> (Nothing, Just $ Right r)-        Just l  -> (Nothing, Just $ Left  l)--{-# INLINABLE signaling #-}-signaling :: IO () -> IO () -> Fetch a -> Fetch a-signaling signalEnd signalElement (Fetch fetchIO) =-  Fetch $ do-    fetch <- fetchIO-    case fetch of-      Nothing      -> signalEnd $> Nothing-      Just element -> signalElement >> return (Just element)--{-# INLINABLE ioFetch #-}-ioFetch :: IO (Fetch a) -> Fetch a-ioFetch fetchIO =-  Fetch $ do-    Fetch fetch <- fetchIO-    fetch+  Fetch $ \ nil just -> atomicModifyIORef' unsentListRef $ \ case+    (!head) : tail -> (tail, just head)+    _ -> ([], nil)

@@ -1,54 +1,40 @@ module Potoki.Core.IO where  import Potoki.Core.Prelude-import Potoki.Core.Types import qualified Potoki.Core.Produce as A import qualified Potoki.Core.Consume as B-import qualified Acquire.IO as C+import qualified Potoki.Core.Transform as C+import qualified Potoki.Core.Fetch as D  -produceAndConsume :: Produce input -> Consume input output -> IO output-produceAndConsume (Produce produceAcquire) (Consume consumeIO) =-  C.acquire produceAcquire consumeIO+produceAndConsume :: A.Produce input -> B.Consume input output -> IO output+produceAndConsume (A.Produce produce) (B.Consume consume) =+  do+    (fetch, kill) <- produce+    consume fetch <* kill -produceAndTransformAndConsume :: Produce input -> Transform input anotherInput -> Consume anotherInput output -> IO output-produceAndTransformAndConsume (Produce produceAcquire) (Transform transformAcquire) (Consume consumeIO) =-  C.acquire (produceAcquire >>= transformAcquire) consumeIO+produceAndTransformAndConsume :: A.Produce input -> C.Transform input anotherInput -> B.Consume anotherInput output -> IO output+produceAndTransformAndConsume (A.Produce produce) (C.Transform transform) (B.Consume consume) =+  do+    (fetch, kill) <- produce+    (transform fetch >>= consume) <* kill -produce :: Produce input -> forall x. IO x -> (input -> IO x) -> IO x-produce (Produce produceAcquire) stop emit =-  C.acquire produceAcquire $ \ (Fetch fetchIO) ->-    fix $ \ doLoop -> do-      fetch <- fetchIO-      case fetch of-        Nothing      -> stop-        Just element -> emit element >> doLoop+produce :: A.Produce input -> forall x. IO x -> (input -> IO x) -> IO x+produce (A.Produce produce) stop emit =+  do+    (D.Fetch fetchIO, kill) <- produce+    fix (\ loop -> join (fetchIO stop (\ element -> emit element >> loop))) <* kill -consume :: IO (Maybe input) -> Consume input output -> IO output-consume fetchIO (Consume consumeIO) =-  consumeIO $ Fetch fetchIO+consume :: (forall x. x -> (input -> x) -> IO x) -> B.Consume input output -> IO output+consume fetch (B.Consume consume) =+  consume (D.Fetch fetch)  {-| Fetch all the elements running the provided handler on them -}-fetchAndHandleAll :: Fetch element -> IO () -> (element -> IO ()) -> IO ()-fetchAndHandleAll (Fetch fetchIO) onEnd onElement =-  fix $ \ doLoop -> do-    fetch <- fetchIO-    case fetch of-      Nothing      -> onEnd-      Just element -> onElement element >> doLoop--{-| Fetch and handle just one element -}-fetchAndHandle :: Fetch element -> IO a -> (element -> IO a) -> IO a-fetchAndHandle (Fetch fetchIO) onEnd onElement =-  do-    fetch <- fetchIO-    case fetch of-      Nothing      -> onEnd-      Just element -> onElement element+fetchAndHandleAll :: D.Fetch element -> IO () -> (element -> IO ()) -> IO ()+fetchAndHandleAll (D.Fetch fetchIO) onEnd onElement =+  fix (\ loop -> join (fetchIO onEnd (\ element -> onElement element >> loop))) -transformList :: Transform a b -> [a] -> IO [b]-transformList transform inputList =-  produceAndTransformAndConsume-    (A.list inputList)-    transform-    (B.list)+{-| Fetch just one element -}+fetch :: D.Fetch element -> IO (Maybe element)+fetch (D.Fetch fetchIO) =+  fetchIO Nothing Just

@@ -12,7 +12,6 @@ import Control.Concurrent as Exports import Control.Exception as Exports import Control.Monad as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)-import Control.Monad.IO.Class as Exports import Control.Monad.Fix as Exports hiding (fix) import Control.Monad.ST as Exports import Data.Bits as Exports@@ -32,11 +31,10 @@ import Data.Ix as Exports import Data.List as Exports hiding (sortOn, isSubsequenceOf, uncons, concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl') import Data.Maybe as Exports-import Data.Monoid as Exports hiding (Last(..), (<>), First(..))+import Data.Monoid as Exports import Data.Ord as Exports import Data.Proxy as Exports import Data.Ratio as Exports-import Data.Semigroup as Exports import Data.STRef as Exports import Data.String as Exports import Data.Traversable as Exports@@ -78,7 +76,3 @@ -- stm ------------------------- import Control.Concurrent.STM as Exports---- acquire---------------------------import Acquire.Acquire as Exports

@@ -1,62 +1,40 @@-module Potoki.Core.Produce-(-  Produce(..),-  list,-  transform,-)-where+module Potoki.Core.Produce where  import Potoki.Core.Prelude-import Potoki.Core.Types import qualified Potoki.Core.Fetch as A  +{-|+Passive producer of elements with support for early termination.++Automates the management of resources.+-}+newtype Produce element =+  Produce (IO (A.Fetch element, IO ()))+ deriving instance Functor Produce  instance Applicative Produce where-  pure x = Produce $ do-    refX <- liftIO (newIORef (Just x))-    return (A.maybeRef refX)-  (<*>) (Produce leftAcquire) (Produce rightAcquire) =-    Produce ((<*>) <$> leftAcquire <*> rightAcquire)+  pure x =+    list [x]+  (<*>) (Produce leftIO) (Produce rightIO) =+    Produce $ do+      (leftFetch, leftKill) <- leftIO+      (rightFetch, rightKill) <- rightIO+      return (leftFetch <*> rightFetch, leftKill >> rightKill)  instance Alternative Produce where   empty =-    Produce (pure empty)-  (<|>) (Produce leftAcquire) (Produce rightAcquire) =-    Produce ((<|>) <$> leftAcquire <*> rightAcquire)--instance Monad Produce where-  return = pure-  (>>=) (Produce (Acquire io1)) k2 =-    Produce $ Acquire $ do-      (fetch1, release1) <- io1-      release2Ref <- newIORef (return ())-      let-        fetch2 input1 =-          case k2 input1 of-            Produce (Acquire io2) ->-              A.ioFetch $ do-                join (readIORef release2Ref)-                (fetch2', release2') <- io2-                writeIORef release2Ref release2'-                return fetch2'-        release3 =-          join (readIORef release2Ref) >> release1-        in return (fetch1 >>= fetch2, release3)--instance MonadIO Produce where-  liftIO io =-    Produce (return (liftIO io))+    Produce (pure (empty, pure ()))+  (<|>) (Produce leftIO) (Produce rightIO) =+    Produce $ do+      (leftFetch, leftKill) <- leftIO+      (rightFetch, rightKill) <- rightIO+      return (leftFetch <|> rightFetch, leftKill >> rightKill)  {-# INLINABLE list #-} list :: [input] -> Produce input-list inputList =-  Produce $ liftIO (A.list <$> newIORef inputList)--{-# INLINE transform #-}-transform :: Transform input output -> Produce input -> Produce output-transform (Transform transformAcquire) (Produce produceAcquire) =+list list =   Produce $ do-    fetch <- produceAcquire-    transformAcquire fetch+    unsentListRef <- newIORef list+    return (A.list unsentListRef, return ())

@@ -1,59 +1,68 @@-module Potoki.Core.Transform-(-  Transform(..),-  consume,-  produce,-  mapFetch,-  executeIO,-  take,-)-where+module Potoki.Core.Transform where -import Potoki.Core.Prelude hiding (take)-import Potoki.Core.Types+import Potoki.Core.Prelude import qualified Potoki.Core.Fetch as A+import qualified Potoki.Core.Consume as C+import qualified Potoki.Core.Produce as D+import qualified Deque as B  +newtype Transform input output =+  Transform (A.Fetch input -> IO (A.Fetch output))+ instance Category Transform where   id =-    Transform (return)-  (.) (Transform left) (Transform right) =-    Transform (left <=< right)+    Transform return+  (.) (Transform leftFetchIO) (Transform rightFetchIO) =+    Transform (leftFetchIO <=< rightFetchIO)  instance Profunctor Transform where-  dimap inputMapping outputMapping (Transform acquire) =-    Transform $ \ oldFetch -> do-      newFetch <- acquire (fmap inputMapping oldFetch)-      return $ fmap outputMapping newFetch+  dimap inputMapping outputMapping (Transform fetchIO) =+    Transform (\ inputFetch -> (fmap . fmap) outputMapping (fetchIO (fmap inputMapping inputFetch)))  instance Choice Transform where-  right' :: Transform a b -> Transform (Either c a) (Either c b)-  right' (Transform rightTransformAcquire) =-    Transform $ \ inFetch -> do-      fetchedLeftMaybeRef <- liftIO $ newIORef Nothing-      Fetch rightFetchIO <- rightTransformAcquire (A.rightHandlingLeft (writeIORef fetchedLeftMaybeRef . Just) inFetch)-      return $ Fetch $ do-          rightFetch <- rightFetchIO-          case rightFetch of-            Nothing    -> do-              fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef-              case fetchedLeftMaybe of-                Nothing          -> return Nothing-                Just fetchedLeft -> do-                  writeIORef fetchedLeftMaybeRef Nothing-                  return $ Just (Left fetchedLeft)-            Just element -> return $ Just (Right element)+  right' (Transform rightTransformIO) =+    Transform $ \ (A.Fetch eitherFetchIO) -> do+      fetchedLeftMaybeRef <- newIORef Nothing+      let+        createRightFetchIO =+          rightTransformIO $ A.Fetch $ \ nil just -> join $ eitherFetchIO (return nil) $ \ case+            Right !rightInput -> return (just rightInput)+            Left !leftInput -> writeIORef fetchedLeftMaybeRef (Just leftInput) $> nil+      rightFetchIORef <- newIORef =<< createRightFetchIO+      return $ A.Fetch $ \ nil just -> do+        A.Fetch rightFetchIO <- readIORef rightFetchIORef+        join $ rightFetchIO+          (do+            fetchedLeftMaybe <- readIORef fetchedLeftMaybeRef+            case fetchedLeftMaybe of+              Just fetchedLeft -> do+                writeIORef fetchedLeftMaybeRef Nothing+                writeIORef rightFetchIORef =<< createRightFetchIO+                return (just (Left fetchedLeft))+              Nothing -> return nil)+          (\ right -> return (just (Right right)))  instance Strong Transform where-  first' (Transform firstTransformAcquire) =-    Transform $ \ inFetch -> do-      cacheRef <- liftIO $ newIORef undefined-      outFetch <- firstTransformAcquire (A.firstCachingSecond cacheRef inFetch)-      return $ A.bothFetchingFirst cacheRef outFetch+  first' (Transform firstTransformIO) =+    Transform $ \ (A.Fetch bothFetchIO) -> do+      secondFetchedDequeRef <- newIORef mempty+      A.Fetch firstFetchIO <-+        firstTransformIO $ A.Fetch $ \ nil just ->+        join $ bothFetchIO (return nil) $ \ (!firstFetched, !secondFetched) -> do+          modifyIORef' secondFetchedDequeRef (B.snoc secondFetched)+          return (just firstFetched)+      return $ A.Fetch $ \ nil just -> join $ firstFetchIO (return nil) $ \ !firstFetched -> do+        secondFetchedDeque <- readIORef secondFetchedDequeRef+        case B.uncons secondFetchedDeque of+          Just (!secondFetched, !secondFetchedDequeTail) -> do+            writeIORef secondFetchedDequeRef secondFetchedDequeTail+            return (just (firstFetched, secondFetched))+          Nothing -> return nil  instance Arrow Transform where   arr fn =-    Transform (return . fmap fn)+    Transform (pure . fmap fn)   first =     first' @@ -62,69 +71,57 @@     left'  {-# INLINE consume #-}-consume :: Consume input output -> Transform input output-consume (Consume runFetch) =-  Transform $ \ (Fetch inputIO) -> do-    stoppedRef <- liftIO $ newIORef False-    return $ Fetch $ do+consume :: C.Consume input output -> Transform input output+consume (C.Consume runFetch) =+  Transform $ \ (A.Fetch fetch) -> do+    stoppedRef <- newIORef False+    return $ A.Fetch $ \ nil just -> do       stopped <- readIORef stoppedRef       if stopped-        then do-          writeIORef stoppedRef False-          return Nothing+        then return nil         else do           emittedRef <- newIORef False-          output <- runFetch $ Fetch $ do-            input <- inputIO-            case input of-              Nothing     -> do-                writeIORef stoppedRef True-                return Nothing-              Just element -> do-                writeIORef emittedRef True-                return $ Just element-          checkStopped <- readIORef stoppedRef-          if checkStopped+          output <-+            runFetch $ A.Fetch $ \ inputNil inputJust ->+            join+              (fetch+                (do+                  writeIORef stoppedRef True+                  return inputNil)+                (\ !input -> do+                  writeIORef emittedRef True+                  return (inputJust input)))+          stopped <- readIORef stoppedRef+          if stopped             then do               emitted <- readIORef emittedRef               if emitted-                then return $ Just output-                else do-                  writeIORef stoppedRef False-                  return Nothing-            else return $ Just output+                then return (just output)+                else return nil+            else return (just output)  {-# INLINABLE produce #-}-produce :: (input -> Produce output) -> Transform input output+produce :: (input -> D.Produce output) -> Transform input output produce inputToProduce =-  Transform $ \ (Fetch inputFetchIO) -> do-    stateRef <- liftIO $ newIORef Nothing-    return $ Fetch $ fix $ \ doLoop -> do+  Transform $ \ (A.Fetch inputFetchIO) -> do+    stateRef <- newIORef Nothing+    return $ A.Fetch $ \ nil just -> fix $ \ loop -> do       state <- readIORef stateRef       case state of-        Just (Fetch outputFetchIO, kill) ->-          do-            outputFetchResult <- outputFetchIO-            case outputFetchResult of-              Just x -> return (Just x)-              Nothing -> do-                kill-                writeIORef stateRef Nothing-                doLoop+        Just (A.Fetch outputFetchIO, kill) ->+          join $ outputFetchIO+            (kill >> writeIORef stateRef Nothing >> loop)+            (return . just)         Nothing ->-          do-            inputFetchResult <- inputFetchIO-            case inputFetchResult of-              Just input -> do-                case inputToProduce input of-                  Produce (Acquire produceIO) -> do-                    fetchAndKill <- produceIO-                    writeIORef stateRef (Just fetchAndKill)-                    doLoop-              Nothing -> return Nothing+          join $ inputFetchIO (return nil) $ \ !input -> do+            case inputToProduce input of+              D.Produce produceIO -> do+                fetchAndKill <- produceIO+                writeIORef stateRef (Just fetchAndKill)+                loop  {-# INLINE mapFetch #-}-mapFetch :: (Fetch a -> Fetch b) -> Transform a b+mapFetch :: (A.Fetch a -> A.Fetch b) -> Transform a b mapFetch mapping =   Transform $ return . mapping @@ -134,21 +131,18 @@ {-# INLINE executeIO #-} executeIO :: Transform (IO a) a executeIO =-  mapFetch $ \ (Fetch fetchIO) -> Fetch (fetchIO >>= sequence)+  mapFetch $ \ (A.Fetch fetchIO) -> A.Fetch $ \ nil just ->+  join (fetchIO (return nil) (fmap just))  {-# INLINE take #-} take :: Int -> Transform input input-take amount-  | amount <= 0 =-    Transform $ \ _ -> return $ Fetch $ return Nothing-  | otherwise   =-    Transform $ \ (Fetch fetchIO) -> do-      countRef <- liftIO $ newIORef amount-      return $ Fetch $ do-        count <- readIORef countRef-        if count > 0-          then do-            modifyIORef countRef pred-            fetchIO-          else-            return Nothing+take amount =+  Transform $ \ (A.Fetch fetchIO) -> do+    countRef <- newIORef amount+    return $ A.Fetch $ \ nil just -> do+      count <- readIORef countRef+      if count > 0+        then do+          writeIORef countRef $! pred count+          fetchIO nil just+        else return nil

@@ -1,34 +0,0 @@-module Potoki.Core.Types-where--import Potoki.Core.Prelude---{-|-Passive producer of elements.--}-newtype Fetch element =-  Fetch (IO (Maybe element))--{-|-Passive producer of elements with support for early termination-and resource management.--}-newtype Produce element =-  Produce (Acquire (Fetch element))--{-|-Active consumer of input into output.-Sort of like a reducer in Map/Reduce.--Automates the management of resources.--}-newtype Consume input output =-  {-|-  An action, which executes the provided fetch in IO,-  while managing the resources behind the scenes.-  -}-  Consume (Fetch input -> IO output)--newtype Transform input output =-  Transform (Fetch input -> Acquire (Fetch output))

@@ -1,7 +1,7 @@ name:   potoki-core version:-  0.12+  1.2 synopsis:   Low-level components of "potoki" description:@@ -11,15 +11,15 @@ category:   Streaming homepage:-  https://github.com/metrix-ai/potoki-core +  https://github.com/nikita-volkov/potoki-core  bug-reports:-  https://github.com/metrix-ai/potoki-core/issues +  https://github.com/nikita-volkov/potoki-core/issues  author:   Nikita Volkov <nikita.y.volkov@mail.ru> maintainer:-  Metrix.AI Ninjas <ninjas@metrix.ai>+  Nikita Volkov <nikita.y.volkov@mail.ru> copyright:-  (c) 2017, Metrix.AI+  (c) 2017, Nikita Volkov license:   MIT license-file:@@ -33,13 +33,13 @@   type:     git   location:-    git://github.com/metrix-ai/potoki-core.git+    git://github.com/nikita-volkov/potoki-core.git  library   hs-source-dirs:     library   default-extensions:-    Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples+    Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples   default-language:     Haskell2010   exposed-modules:@@ -49,13 +49,15 @@     Potoki.Core.IO     Potoki.Core.Transform   other-modules:-    Potoki.Core.Types     Potoki.Core.Prelude   build-depends:-    acquire >=0.2 && <0.3,-    base >=4.7 && <5,+    -- +    deque >=0.2 && <0.3,+    -- +    stm >=2.4 && <3,+    --      profunctors >=5.2 && <6,-    stm >=2.4 && <3+    base >=4.7 && <5  test-suite tests   type:@@ -65,7 +67,7 @@   main-is:     Main.hs   default-extensions:-    Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples+    Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples   default-language:     Haskell2010   build-depends:

@@ -21,8 +21,8 @@     testProperty "list to list" $ \ (list :: [Int]) ->     list === unsafePerformIO (C.produceAndConsume (E.list list) D.list)     ,-    testProperty "consecutive consumers" $ \ (list :: [Int], amount) ->-    list === unsafePerformIO (C.produceAndConsume (E.list list) ((++) <$> D.transform (A.take amount) D.list <*> D.list))+    testProperty "Dual consumption" $ \ (list :: [Int]) ->+    (list, list) === unsafePerformIO (C.produceAndConsume (E.list list) ((,) <$> D.list <*> D.list))     ,     transform   ]@@ -38,35 +38,13 @@   ]  transformProduce =-  testGroup "Produce" $ -  [-    testCase "1" $ do-      let-        list = [1, 2, 3] :: [Int]-      result <- C.produceAndTransformAndConsume-        (E.list list)-        (A.produce (E.list . \ n -> flip replicate n n))-        (D.list)-      assertEqual "" [1, 2, 2, 3, 3, 3] result-  ,-    testCase "2" $ do-      let-        list = [1, 2, 3] :: [Int]-      result <- C.produceAndTransformAndConsume-        (E.list list)-        (A.produce (E.list . \ n -> [(n, n)]))-        (D.list)-      assertEqual "" [(1, 1), (2, 2), (3, 3)] result-  ,-    testCase "3" $ do-      let-        list = [1, 2, 3] :: [Int]-      result <- C.produceAndTransformAndConsume-        (E.list list)-        (A.produce (E.list . \ n -> [n, n]))-        (D.list)-      assertEqual "" [1, 1, 2, 2, 3, 3] result-  ]+  testCase "Produce" $ do+    let list = [1, 2, 3] :: [Int]+    result <- C.produceAndTransformAndConsume+      (E.list list)+      (A.produce (E.list . \ n -> flip replicate n n))+      (D.list)+    assertEqual "" [1, 2, 2, 3, 3, 3] result  transformChoice =   testGroup "Choice" $@@ -81,38 +59,21 @@     testCase "2" $ do       let         list = [Left 1, Left 2, Right 'z', Right 'a', Right 'b', Left 0, Right 'x', Left 4, Left 3]-        transform = right' (A.consume D.list)+        transform = right (A.consume D.list)       result <- C.produceAndTransformAndConsume (E.list list) transform D.list       assertEqual "" [Left 1, Left 2, Right "zab", Left 0, Right "x", Left 4, Left 3] result     ,     testCase "3" $ do       let-        list = [Left 4, Right 'z', Right 'a', Left 3, Right 'b', Left 0, Left 1, Right 'x', Left 4, Left 3]-        transform = left' (A.consume D.list)+        list = [Right 'z', Right 'a', Left 3, Right 'b', Left 0, Left 1, Right 'x', Left 4, Left 3]+        transform = left (A.consume D.list)       result <- C.produceAndTransformAndConsume (E.list list) transform D.list-      assertEqual "" [Left [4], Right 'z', Right 'a', Left [3], Right 'b', Left [0, 1], Right 'x', Left [4, 3]] result+      assertEqual "" [Right 'z', Right 'a', Left [3], Right 'b', Left [0, 1], Right 'x', Left [4, 3]] result   ]  transformArrowLaws =   testGroup "Arrow laws"   [-    testGroup "Strong"-    [-      testCase "1" $ do-        let-          input = [(1,'a'),(2,'b'),(3,'c'),(4,'d')]-          transform = first transform1-        result <- C.produceAndTransformAndConsume (E.list input) transform D.list-        assertEqual "" [(6,'c'),(4,'d')] result-      ,-      testCase "Lack of elements" $ do-        let-          input = [(1,'a'),(2,'b')]-          transform = first transform1-        result <- C.produceAndTransformAndConsume (E.list input) transform D.list-        assertEqual "" [(3,'b')] result-    ]-    ,     transformProperty "arr id = id"       (arr id :: A.Transform Int Int)       id@@ -168,8 +129,8 @@   where     f = (+24) :: Int -> Int     g = (*3) :: Int -> Int-    transform1 = A.consume (D.transform (A.take 3) D.sum) :: A.Transform Int Int-    transform2 = A.consume (D.transform (A.take 4) D.sum) :: A.Transform Int Int+    transform1 = A.take 3 >>> A.consume D.sum :: A.Transform Int Int+    transform2 = A.take 4 >>> A.consume D.sum :: A.Transform Int Int     assoc ((a,b),c) = (a,(b,c))     assocsum (Left (Left x)) = Left x     assocsum (Left (Right y)) = Right (Left y)