{-# LANGUAGE FlexibleContexts #-}

-- |
-- Module     : Simulation.Aivika.Trans.Processor
-- Copyright  : Copyright (c) 2009-2017, David Sorokin <david.sorokin@gmail.com>
-- License    : BSD3
-- Maintainer : David Sorokin <david.sorokin@gmail.com>
-- Stability  : experimental
-- Tested with: GHC 8.0.1
--
-- The processor of simulation data.
--
module Simulation.Aivika.Trans.Processor
       (-- * Processor Type
        Processor(..),
        -- * Processor Primitives
        emptyProcessor,
        arrProcessor,
        accumProcessor,
        withinProcessor,
        -- * Specifying Identifier
        processorUsingId,
        -- * Prefetch and Delay Processors
        prefetchProcessor,
        delayProcessor,
        -- * Buffer Processor
        bufferProcessor,
        bufferProcessorLoop,
        -- * Processing Queues
        queueProcessor,
        queueProcessorLoopMerging,
        queueProcessorLoopSeq,
        queueProcessorLoopParallel,
        -- * Sequencing Processors
        processorSeq,
        -- * Parallelizing Processors
        processorParallel,
        processorQueuedParallel,
        processorPrioritisingOutputParallel,
        processorPrioritisingInputParallel,
        processorPrioritisingInputOutputParallel,
        -- * Arrival Processor
        arrivalProcessor,
        -- * Utilities
        joinProcessor,
        -- * Failover
        failoverProcessor,
        -- * Integrating with Signals and Channels
        channelProcessor,
        processorChannel,
        queuedChannelProcessor,
        queuedProcessorChannel,
        -- * Debugging
        traceProcessor) where

import qualified Control.Category as C
import Control.Arrow

import Data.Monoid

import Simulation.Aivika.Trans.DES
import Simulation.Aivika.Trans.Simulation
import Simulation.Aivika.Trans.Dynamics
import Simulation.Aivika.Trans.Event
import Simulation.Aivika.Trans.Composite
import Simulation.Aivika.Trans.Cont
import Simulation.Aivika.Trans.Process
import Simulation.Aivika.Trans.Stream
import Simulation.Aivika.Trans.QueueStrategy
import Simulation.Aivika.Trans.Signal
import Simulation.Aivika.Trans.Channel
import Simulation.Aivika.Arrival (Arrival(..))

-- | Represents a processor of simulation data.
newtype Processor m a b =
  Processor { forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor :: Stream m a -> Stream m b
              -- ^ Run the processor.
            }

instance C.Category (Processor m) where

  {-# INLINE id #-}
  id :: forall a. Processor m a a
id  = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a. a -> a
id

  {-# INLINE (.) #-}
  Processor Stream m b -> Stream m c
x . :: forall b c a. Processor m b c -> Processor m a b -> Processor m a c
. Processor Stream m a -> Stream m b
y = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor (Stream m b -> Stream m c
x forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream m a -> Stream m b
y)

-- The implementation is based on article
-- A New Notation for Arrows by Ross Paterson,
-- although my streams are different and they
-- already depend on the Process monad,
-- while the pure streams were considered in the
-- mentioned article.
  
instance MonadDES m => Arrow (Processor m) where

  {-# INLINABLE arr #-}
  arr :: forall b c. (b -> c) -> Processor m b c
arr = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) a b.
MonadDES m =>
(a -> b) -> Stream m a -> Stream m b
mapStream

  {-# INLINABLE first #-}
  first :: forall b c d. Processor m b c -> Processor m (b, d) (c, d)
first (Processor Stream m b -> Stream m c
f) =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m (b, d)
xys ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do (Stream m b
xs, Stream m d
ys) <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m (a, b) -> Simulation m (Stream m a, Stream m b)
unzipStream Stream m (b, d)
xys
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m a -> Stream m b -> Stream m (a, b)
zipStreamParallel (Stream m b -> Stream m c
f Stream m b
xs) Stream m d
ys

  {-# INLINABLE second #-}
  second :: forall b c d. Processor m b c -> Processor m (d, b) (d, c)
second (Processor Stream m b -> Stream m c
f) =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m (d, b)
xys ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do (Stream m d
xs, Stream m b
ys) <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m (a, b) -> Simulation m (Stream m a, Stream m b)
unzipStream Stream m (d, b)
xys
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m a -> Stream m b -> Stream m (a, b)
zipStreamParallel Stream m d
xs (Stream m b -> Stream m c
f Stream m b
ys)

  {-# INLINABLE (***) #-}
  Processor Stream m b -> Stream m c
f *** :: forall b c b' c'.
Processor m b c -> Processor m b' c' -> Processor m (b, b') (c, c')
*** Processor Stream m b' -> Stream m c'
g =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m (b, b')
xys ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do (Stream m b
xs, Stream m b'
ys) <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m (a, b) -> Simulation m (Stream m a, Stream m b)
unzipStream Stream m (b, b')
xys
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m a -> Stream m b -> Stream m (a, b)
zipStreamParallel (Stream m b -> Stream m c
f Stream m b
xs) (Stream m b' -> Stream m c'
g Stream m b'
ys)

  {-# INLINABLE (&&&) #-}
  Processor Stream m b -> Stream m c
f &&& :: forall b c c'.
Processor m b c -> Processor m b c' -> Processor m b (c, c')
&&& Processor Stream m b -> Stream m c'
g =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m b
xs -> forall (m :: * -> *) a b.
MonadDES m =>
Stream m a -> Stream m b -> Stream m (a, b)
zipStreamParallel (Stream m b -> Stream m c
f Stream m b
xs) (Stream m b -> Stream m c'
g Stream m b
xs)

instance MonadDES m => ArrowChoice (Processor m) where

  {-# INLINABLE left #-}
  left :: forall b c d.
Processor m b c -> Processor m (Either b d) (Either c d)
left (Processor Stream m b -> Stream m c
f) =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m (Either b d)
xs ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do Stream m (Either b d)
ys <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Simulation m (Stream m a)
memoStream Stream m (Either b d)
xs
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b c.
MonadDES m =>
Stream m (Either a b) -> Stream m c -> Stream m (Either c b)
replaceLeftStream Stream m (Either b d)
ys (Stream m b -> Stream m c
f forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m (Either a b) -> Stream m a
leftStream Stream m (Either b d)
ys)

  {-# INLINABLE right #-}
  right :: forall b c d.
Processor m b c -> Processor m (Either d b) (Either d c)
right (Processor Stream m b -> Stream m c
f) =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m (Either d b)
xs ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do Stream m (Either d b)
ys <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Simulation m (Stream m a)
memoStream Stream m (Either d b)
xs
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b c.
MonadDES m =>
Stream m (Either a b) -> Stream m c -> Stream m (Either a c)
replaceRightStream Stream m (Either d b)
ys (Stream m b -> Stream m c
f forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
MonadDES m =>
Stream m (Either a b) -> Stream m b
rightStream Stream m (Either d b)
ys)

instance MonadDES m => ArrowZero (Processor m) where

  {-# INLINE zeroArrow #-}
  zeroArrow :: forall b c. Processor m b c
zeroArrow = forall (m :: * -> *) a b. MonadDES m => Processor m a b
emptyProcessor

instance MonadDES m => ArrowPlus (Processor m) where

  {-# INLINABLE (<+>) #-}
  (Processor Stream m b -> Stream m c
f) <+> :: forall b c. Processor m b c -> Processor m b c -> Processor m b c
<+> (Processor Stream m b -> Stream m c
g) =
    forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m b
xs ->
    forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
    do [Stream m b
xs1, Stream m b
xs2] <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a.
MonadDES m =>
Int -> Stream m a -> Simulation m [Stream m a]
splitStream Int
2 Stream m b
xs
       forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Stream m a -> Stream m a
mergeStreams (Stream m b -> Stream m c
f Stream m b
xs1) (Stream m b -> Stream m c
g Stream m b
xs2)

-- | A processor that never finishes its work producing an 'emptyStream'.
emptyProcessor :: MonadDES m => Processor m a b
{-# INLINABLE emptyProcessor #-}
emptyProcessor :: forall (m :: * -> *) a b. MonadDES m => Processor m a b
emptyProcessor = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ forall a b. a -> b -> a
const forall (m :: * -> *) a. MonadDES m => Stream m a
emptyStream

-- | Create a simple processor by the specified handling function
-- that runs the discontinuous process for each input value to get the output.
arrProcessor :: MonadDES m => (a -> Process m b) -> Processor m a b
{-# INLINABLE arrProcessor #-}
arrProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
(a -> Process m b) -> Processor m a b
arrProcessor = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) a b.
MonadDES m =>
(a -> Process m b) -> Stream m a -> Stream m b
mapStreamM

-- | Accumulator that outputs a value determined by the supplied function.
accumProcessor :: MonadDES m => (acc -> a -> Process m (acc, b)) -> acc -> Processor m a b
{-# INLINABLE accumProcessor #-}
accumProcessor :: forall (m :: * -> *) acc a b.
MonadDES m =>
(acc -> a -> Process m (acc, b)) -> acc -> Processor m a b
accumProcessor acc -> a -> Process m (acc, b)
f acc
acc =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs -> forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$ Stream m a -> acc -> Process m (b, Stream m b)
loop Stream m a
xs acc
acc where
    loop :: Stream m a -> acc -> Process m (b, Stream m b)
loop Stream m a
xs acc
acc =
      do (a
a, Stream m a
xs') <- forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m a
xs
         (acc
acc', b
b) <- acc -> a -> Process m (acc, b)
f acc
acc a
a
         forall (m :: * -> *) a. Monad m => a -> m a
return (b
b, forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$ Stream m a -> acc -> Process m (b, Stream m b)
loop Stream m a
xs' acc
acc') 

-- | Involve the computation with side effect when processing a stream of data.
withinProcessor :: MonadDES m => Process m () -> Processor m a a
{-# INLINABLE withinProcessor #-}
withinProcessor :: forall (m :: * -> *) a.
MonadDES m =>
Process m () -> Processor m a a
withinProcessor Process m ()
m =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$
  forall (m :: * -> *) a b.
MonadDES m =>
(a -> Process m b) -> Stream m a -> Stream m b
mapStreamM forall a b. (a -> b) -> a -> b
$ \a
a ->
  do { Process m ()
m; forall (m :: * -> *) a. Monad m => a -> m a
return a
a }

-- | Create a processor that will use the specified process identifier.
-- It can be useful to refer to the underlying 'Process' computation which
-- can be passivated, interrupted, canceled and so on. See also the
-- 'processUsingId' function for more details.
processorUsingId :: MonadDES m => ProcessId m -> Processor m a b -> Processor m a b
{-# INLINABLE processorUsingId #-}
processorUsingId :: forall (m :: * -> *) a b.
MonadDES m =>
ProcessId m -> Processor m a b -> Processor m a b
processorUsingId ProcessId m
pid (Processor Stream m a -> Stream m b
f) =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) a.
MonadDES m =>
ProcessId m -> Process m a -> Process m a
processUsingId ProcessId m
pid forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream m a -> Stream m b
f

-- | Launches the specified processors in parallel consuming the same input
-- stream and producing a combined output stream.
--
-- If you don't know what the enqueue strategies to apply, then
-- you will probably need 'FCFS' for the both parameters, or
-- function 'processorParallel' that does namely this.
processorQueuedParallel :: (MonadDES m,
                            EnqueueStrategy m si,
                            EnqueueStrategy m so)
                           => si
                           -- ^ the strategy applied for enqueuing the input data
                           -> so
                           -- ^ the strategy applied for enqueuing the output data
                           -> [Processor m a b]
                           -- ^ the processors to parallelize
                           -> Processor m a b
                           -- ^ the parallelized processor
{-# INLINABLE processorQueuedParallel #-}
processorQueuedParallel :: forall (m :: * -> *) si so a b.
(MonadDES m, EnqueueStrategy m si, EnqueueStrategy m so) =>
si -> so -> [Processor m a b] -> Processor m a b
processorQueuedParallel si
si so
so [Processor m a b]
ps =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do let n :: Int
n = forall (t :: * -> *) a. Foldable t => t a -> Int
length [Processor m a b]
ps
     [Stream m a]
input <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) s a.
(MonadDES m, EnqueueStrategy m s) =>
s -> Int -> Stream m a -> Simulation m [Stream m a]
splitStreamQueueing si
si Int
n Stream m a
xs
     let results :: [Stream m b]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream m a]
input [Processor m a b]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream m a
input, Processor m a b
p) ->
           forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m a b
p Stream m a
input
         output :: Stream m b
output  = forall (m :: * -> *) s a.
(MonadDES m, EnqueueStrategy m s) =>
s -> [Stream m a] -> Stream m a
concatQueuedStreams so
so [Stream m b]
results
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Launches the specified processors in parallel using priorities for combining the output.
processorPrioritisingOutputParallel :: (MonadDES m,
                                        EnqueueStrategy m si,
                                        PriorityQueueStrategy m so po)
                                       => si
                                       -- ^ the strategy applied for enqueuing the input data
                                       -> so
                                       -- ^ the strategy applied for enqueuing the output data
                                       -> [Processor m a (po, b)]
                                       -- ^ the processors to parallelize
                                       -> Processor m a b
                                       -- ^ the parallelized processor
{-# INLINABLE processorPrioritisingOutputParallel #-}
processorPrioritisingOutputParallel :: forall (m :: * -> *) si so po a b.
(MonadDES m, EnqueueStrategy m si,
 PriorityQueueStrategy m so po) =>
si -> so -> [Processor m a (po, b)] -> Processor m a b
processorPrioritisingOutputParallel si
si so
so [Processor m a (po, b)]
ps =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do let n :: Int
n = forall (t :: * -> *) a. Foldable t => t a -> Int
length [Processor m a (po, b)]
ps
     [Stream m a]
input <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) s a.
(MonadDES m, EnqueueStrategy m s) =>
s -> Int -> Stream m a -> Simulation m [Stream m a]
splitStreamQueueing si
si Int
n Stream m a
xs
     let results :: [Stream m (po, b)]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream m a]
input [Processor m a (po, b)]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream m a
input, Processor m a (po, b)
p) ->
           forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m a (po, b)
p Stream m a
input
         output :: Stream m b
output  = forall (m :: * -> *) s p a.
(MonadDES m, PriorityQueueStrategy m s p) =>
s -> [Stream m (p, a)] -> Stream m a
concatPriorityStreams so
so [Stream m (po, b)]
results
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Launches the specified processors in parallel using priorities for consuming the intput.
processorPrioritisingInputParallel :: (MonadDES m,
                                       PriorityQueueStrategy m si pi,
                                       EnqueueStrategy m so)
                                      => si
                                      -- ^ the strategy applied for enqueuing the input data
                                      -> so
                                      -- ^ the strategy applied for enqueuing the output data
                                      -> [(Stream m pi, Processor m a b)]
                                      -- ^ the streams of input priorities and the processors
                                      -- to parallelize
                                      -> Processor m a b
                                      -- ^ the parallelized processor
{-# INLINABLE processorPrioritisingInputParallel #-}
processorPrioritisingInputParallel :: forall (m :: * -> *) si pi so a b.
(MonadDES m, PriorityQueueStrategy m si pi,
 EnqueueStrategy m so) =>
si -> so -> [(Stream m pi, Processor m a b)] -> Processor m a b
processorPrioritisingInputParallel si
si so
so [(Stream m pi, Processor m a b)]
ps =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do [Stream m a]
input <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) s p a.
(MonadDES m, PriorityQueueStrategy m s p) =>
s -> [Stream m p] -> Stream m a -> Simulation m [Stream m a]
splitStreamPrioritising si
si (forall a b. (a -> b) -> [a] -> [b]
map forall a b. (a, b) -> a
fst [(Stream m pi, Processor m a b)]
ps) Stream m a
xs
     let results :: [Stream m b]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream m a]
input [(Stream m pi, Processor m a b)]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream m a
input, (Stream m pi
_, Processor m a b
p)) ->
           forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m a b
p Stream m a
input
         output :: Stream m b
output  = forall (m :: * -> *) s a.
(MonadDES m, EnqueueStrategy m s) =>
s -> [Stream m a] -> Stream m a
concatQueuedStreams so
so [Stream m b]
results
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Launches the specified processors in parallel using priorities for consuming
-- the input and combining the output.
processorPrioritisingInputOutputParallel :: (MonadDES m,
                                             PriorityQueueStrategy m si pi,
                                             PriorityQueueStrategy m so po)
                                            => si
                                            -- ^ the strategy applied for enqueuing the input data
                                            -> so
                                            -- ^ the strategy applied for enqueuing the output data
                                            -> [(Stream m pi, Processor m a (po, b))]
                                            -- ^ the streams of input priorities and the processors
                                            -- to parallelize
                                            -> Processor m a b
                                            -- ^ the parallelized processor
{-# INLINABLE processorPrioritisingInputOutputParallel #-}
processorPrioritisingInputOutputParallel :: forall (m :: * -> *) si pi so po a b.
(MonadDES m, PriorityQueueStrategy m si pi,
 PriorityQueueStrategy m so po) =>
si
-> so -> [(Stream m pi, Processor m a (po, b))] -> Processor m a b
processorPrioritisingInputOutputParallel si
si so
so [(Stream m pi, Processor m a (po, b))]
ps =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do [Stream m a]
input <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) s p a.
(MonadDES m, PriorityQueueStrategy m s p) =>
s -> [Stream m p] -> Stream m a -> Simulation m [Stream m a]
splitStreamPrioritising si
si (forall a b. (a -> b) -> [a] -> [b]
map forall a b. (a, b) -> a
fst [(Stream m pi, Processor m a (po, b))]
ps) Stream m a
xs
     let results :: [Stream m (po, b)]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream m a]
input [(Stream m pi, Processor m a (po, b))]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream m a
input, (Stream m pi
_, Processor m a (po, b)
p)) ->
           forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m a (po, b)
p Stream m a
input
         output :: Stream m b
output  = forall (m :: * -> *) s p a.
(MonadDES m, PriorityQueueStrategy m s p) =>
s -> [Stream m (p, a)] -> Stream m a
concatPriorityStreams so
so [Stream m (po, b)]
results
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Launches the processors in parallel consuming the same input stream and producing
-- a combined output stream. This version applies the 'FCFS' strategy both for input
-- and output, which suits the most part of uses cases.
processorParallel :: MonadDES m => [Processor m a b] -> Processor m a b
{-# INLINABLE processorParallel #-}
processorParallel :: forall (m :: * -> *) a b.
MonadDES m =>
[Processor m a b] -> Processor m a b
processorParallel = forall (m :: * -> *) si so a b.
(MonadDES m, EnqueueStrategy m si, EnqueueStrategy m so) =>
si -> so -> [Processor m a b] -> Processor m a b
processorQueuedParallel FCFS
FCFS FCFS
FCFS

-- | Launches the processors sequentially using the 'prefetchProcessor' between them
-- to model an autonomous work of each of the processors specified.
processorSeq :: MonadDES m => [Processor m a a] -> Processor m a a
{-# INLINABLE processorSeq #-}
processorSeq :: forall (m :: * -> *) a.
MonadDES m =>
[Processor m a a] -> Processor m a a
processorSeq []  = forall (m :: * -> *) a b. MonadDES m => Processor m a b
emptyProcessor
processorSeq [Processor m a a
p] = Processor m a a
p
processorSeq (Processor m a a
p : [Processor m a a]
ps) = Processor m a a
p forall {k} (cat :: k -> k -> *) (a :: k) (b :: k) (c :: k).
Category cat =>
cat a b -> cat b c -> cat a c
>>> forall (m :: * -> *) a. MonadDES m => Processor m a a
prefetchProcessor forall {k} (cat :: k -> k -> *) (a :: k) (b :: k) (c :: k).
Category cat =>
cat a b -> cat b c -> cat a c
>>> forall (m :: * -> *) a.
MonadDES m =>
[Processor m a a] -> Processor m a a
processorSeq [Processor m a a]
ps

-- | Create a buffer processor, where the process from the first argument
-- consumes the input stream but the stream passed in as the second argument
-- and produced usually by some other process is returned as an output.
-- This kind of processor is very useful for modeling the queues.
bufferProcessor :: MonadDES m
                   => (Stream m a -> Process m ())
                   -- ^ a separate process to consume the input 
                   -> Stream m b
                   -- ^ the resulting stream of data
                   -> Processor m a b
{-# INLINABLE bufferProcessor #-}
bufferProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
(Stream m a -> Process m ()) -> Stream m b -> Processor m a b
bufferProcessor Stream m a -> Process m ()
consume Stream m b
output =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do forall (m :: * -> *). MonadDES m => Process m () -> Process m ()
spawnProcess (Stream m a -> Process m ()
consume Stream m a
xs)
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Like 'bufferProcessor' but allows creating a loop when some items
-- can be processed repeatedly. It is very useful for modeling the processors 
-- with queues and loop-backs.
bufferProcessorLoop :: MonadDES m
                       => (Stream m a -> Stream m c -> Process m ())
                       -- ^ consume two streams: the input values of type @a@
                       -- and the values of type @c@ returned by the loop
                       -> Stream m d
                       -- ^ the stream of data that may become results
                       -> Processor m d (Either e b)
                       -- ^ process and then decide what values of type @e@
                       -- should be processed in the loop (this is a condition)
                       -> Processor m e c
                       -- ^ process in the loop and then return a value
                       -- of type @c@ to the input again (this is a loop body)
                       -> Processor m a b
{-# INLINABLE bufferProcessorLoop #-}
bufferProcessorLoop :: forall (m :: * -> *) a c d e b.
MonadDES m =>
(Stream m a -> Stream m c -> Process m ())
-> Stream m d
-> Processor m d (Either e b)
-> Processor m e c
-> Processor m a b
bufferProcessorLoop Stream m a -> Stream m c -> Process m ()
consume Stream m d
preoutput Processor m d (Either e b)
cond Processor m e c
body =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do (Stream m e
reverted, Stream m b
output) <-
       forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
SimulationLift t m =>
Simulation m a -> t m a
liftSimulation forall a b. (a -> b) -> a -> b
$
       forall (m :: * -> *) a b.
MonadDES m =>
Stream m (Either a b) -> Simulation m (Stream m a, Stream m b)
partitionEitherStream forall a b. (a -> b) -> a -> b
$
       forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m d (Either e b)
cond Stream m d
preoutput
     forall (m :: * -> *). MonadDES m => Process m () -> Process m ()
spawnProcess
       (Stream m a -> Stream m c -> Process m ()
consume Stream m a
xs forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m e c
body Stream m e
reverted)
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
output

-- | Return a processor with help of which we can model the queue.
--
-- Although the function doesn't refer to the queue directly, its main use case
-- is namely a processing of the queue. The first argument should be the enqueueing
-- operation, while the second argument should be the opposite dequeueing operation.
--
-- The reason is as follows. There are many possible combinations how the queues
-- can be modeled. There is no sense to enumerate all them creating a separate function
-- for each case. We can just use combinators to define exactly what we need.
--
-- So, the queue can lose the input items if the queue is full, or the input process
-- can suspend while the queue is full, or we can use priorities for enqueueing,
-- storing and dequeueing the items in different combinations. There are so many use
-- cases!
--
-- There is a hope that this function along with other similar functions from this
-- module is sufficient to cover the most important cases. Even if it is not sufficient
-- then you can use a more generic function 'bufferProcessor' which this function is
-- based on. In case of need, you can even write your own function from scratch. It is
-- quite easy actually.
queueProcessor :: MonadDES m =>
                  (a -> Process m ())
                  -- ^ enqueue the input item and wait
                  -- while the queue is full if required
                  -- so that there were no hanging items
                  -> Process m b
                  -- ^ dequeue an output item
                  -> Processor m a b
                  -- ^ the buffering processor
{-# INLINABLE queueProcessor #-}
queueProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
(a -> Process m ()) -> Process m b -> Processor m a b
queueProcessor a -> Process m ()
enqueue Process m b
dequeue =
  forall (m :: * -> *) a b.
MonadDES m =>
(Stream m a -> Process m ()) -> Stream m b -> Processor m a b
bufferProcessor
  (forall (m :: * -> *) a.
MonadDES m =>
(a -> Process m ()) -> Stream m a -> Process m ()
consumeStream a -> Process m ()
enqueue)
  (forall (m :: * -> *) a. MonadDES m => Process m a -> Stream m a
repeatProcess Process m b
dequeue)

-- | Like 'queueProcessor' creates a queue processor but with a loop when some items 
-- can be processed and then added to the queue again. Also it allows specifying 
-- how two input streams of data can be merged.
queueProcessorLoopMerging :: MonadDES m
                             => (Stream m a -> Stream m d -> Stream m e)
                             -- ^ merge two streams: the input values of type @a@
                             -- and the values of type @d@ returned by the loop
                             -> (e -> Process m ())
                             -- ^ enqueue the input item and wait
                             -- while the queue is full if required
                             -- so that there were no hanging items
                             -> Process m c
                             -- ^ dequeue an item for the further processing
                             -> Processor m c (Either f b)
                             -- ^ process and then decide what values of type @f@
                             -- should be processed in the loop (this is a condition)
                             -> Processor m f d
                             -- ^ process in the loop and then return a value
                             -- of type @d@ to the queue again (this is a loop body)
                             -> Processor m a b
                             -- ^ the buffering processor
{-# INLINABLE queueProcessorLoopMerging #-}
queueProcessorLoopMerging :: forall (m :: * -> *) a d e c f b.
MonadDES m =>
(Stream m a -> Stream m d -> Stream m e)
-> (e -> Process m ())
-> Process m c
-> Processor m c (Either f b)
-> Processor m f d
-> Processor m a b
queueProcessorLoopMerging Stream m a -> Stream m d -> Stream m e
merge e -> Process m ()
enqueue Process m c
dequeue =
  forall (m :: * -> *) a c d e b.
MonadDES m =>
(Stream m a -> Stream m c -> Process m ())
-> Stream m d
-> Processor m d (Either e b)
-> Processor m e c
-> Processor m a b
bufferProcessorLoop
  (\Stream m a
bs Stream m d
cs ->
    forall (m :: * -> *) a.
MonadDES m =>
(a -> Process m ()) -> Stream m a -> Process m ()
consumeStream e -> Process m ()
enqueue forall a b. (a -> b) -> a -> b
$
    Stream m a -> Stream m d -> Stream m e
merge Stream m a
bs Stream m d
cs)
  (forall (m :: * -> *) a. MonadDES m => Process m a -> Stream m a
repeatProcess Process m c
dequeue)

-- | Like 'queueProcessorLoopMerging' creates a queue processor with a loop when
-- some items can be processed and then added to the queue again. Only it sequentially 
-- merges two input streams of data: one stream that come from the external source and 
-- another stream of data returned by the loop. The first stream has a priority over 
-- the second one.
queueProcessorLoopSeq :: MonadDES m
                         => (a -> Process m ())
                         -- ^ enqueue the input item and wait
                         -- while the queue is full if required
                         -- so that there were no hanging items
                         -> Process m c
                         -- ^ dequeue an item for the further processing
                         -> Processor m c (Either e b)
                         -- ^ process and then decide what values of type @e@
                         -- should be processed in the loop (this is a condition)
                         -> Processor m e a
                         -- ^ process in the loop and then return a value
                         -- of type @a@ to the queue again (this is a loop body)
                         -> Processor m a b
                         -- ^ the buffering processor
{-# INLINABLE queueProcessorLoopSeq #-}
queueProcessorLoopSeq :: forall (m :: * -> *) a c e b.
MonadDES m =>
(a -> Process m ())
-> Process m c
-> Processor m c (Either e b)
-> Processor m e a
-> Processor m a b
queueProcessorLoopSeq =
  forall (m :: * -> *) a d e c f b.
MonadDES m =>
(Stream m a -> Stream m d -> Stream m e)
-> (e -> Process m ())
-> Process m c
-> Processor m c (Either f b)
-> Processor m f d
-> Processor m a b
queueProcessorLoopMerging forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Stream m a -> Stream m a
mergeStreams

-- | Like 'queueProcessorLoopMerging' creates a queue processor with a loop when
-- some items can be processed and then added to the queue again. Only it runs two 
-- simultaneous processes to enqueue the input streams of data: one stream that come 
-- from the external source and another stream of data returned by the loop.
queueProcessorLoopParallel :: MonadDES m
                              => (a -> Process m ())
                              -- ^ enqueue the input item and wait
                              -- while the queue is full if required
                              -- so that there were no hanging items
                              -> Process m c
                              -- ^ dequeue an item for the further processing
                              -> Processor m c (Either e b)
                              -- ^ process and then decide what values of type @e@
                              -- should be processed in the loop (this is a condition)
                              -> Processor m e a
                              -- ^ process in the loop and then return a value
                              -- of type @a@ to the queue again (this is a loop body)
                              -> Processor m a b
                              -- ^ the buffering processor
{-# INLINABLE queueProcessorLoopParallel #-}
queueProcessorLoopParallel :: forall (m :: * -> *) a c e b.
MonadDES m =>
(a -> Process m ())
-> Process m c
-> Processor m c (Either e b)
-> Processor m e a
-> Processor m a b
queueProcessorLoopParallel a -> Process m ()
enqueue Process m c
dequeue =
  forall (m :: * -> *) a c d e b.
MonadDES m =>
(Stream m a -> Stream m c -> Process m ())
-> Stream m d
-> Processor m d (Either e b)
-> Processor m e c
-> Processor m a b
bufferProcessorLoop
  (\Stream m a
bs Stream m a
cs ->
    do forall (m :: * -> *). MonadDES m => Process m () -> Process m ()
spawnProcess forall a b. (a -> b) -> a -> b
$
         forall (m :: * -> *) a.
MonadDES m =>
(a -> Process m ()) -> Stream m a -> Process m ()
consumeStream a -> Process m ()
enqueue Stream m a
bs
       forall (m :: * -> *). MonadDES m => Process m () -> Process m ()
spawnProcess forall a b. (a -> b) -> a -> b
$
         forall (m :: * -> *) a.
MonadDES m =>
(a -> Process m ()) -> Stream m a -> Process m ()
consumeStream a -> Process m ()
enqueue Stream m a
cs)
  (forall (m :: * -> *) a. MonadDES m => Process m a -> Stream m a
repeatProcess Process m c
dequeue)

-- | This is a prefetch processor that requests for one more data item from 
-- the input in advance while the latest item is not yet fully processed in 
-- the chain of streams, usually by other processors.
--
-- You can think of this as the prefetched processor could place its latest 
-- data item in some temporary space for later use, which is very useful 
-- for modeling a sequence of separate and independent work places.
prefetchProcessor :: MonadDES m => Processor m a a
{-# INLINABLE prefetchProcessor #-}
prefetchProcessor :: forall (m :: * -> *) a. MonadDES m => Processor m a a
prefetchProcessor = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall (m :: * -> *) a. MonadDES m => Stream m a -> Stream m a
prefetchStream

-- | Convert the specified signal transform, i.e. the channel, to a processor.
--
-- The processor may return data with delay as the values are requested by demand.
-- Consider using the 'arrivalSignal' function to provide with the information
-- about the time points at which the signal was actually triggered.
--
-- The point is that the 'Stream' used in the 'Processor' is requested outside, 
-- while the 'Signal' used in the 'Channel' is triggered inside. They are different by nature. 
-- The former is passive, while the latter is active.
--
-- The resulting processor may be a root of space leak as it uses an internal queue to store
-- the values received from the input signal. Consider using 'queuedChannelProcessor' that
-- allows specifying the bounded queue in case of need.
channelProcessor :: MonadDES m => Channel m a b -> Processor m a b
{-# INLINABLE channelProcessor #-}
channelProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
Channel m a b -> Processor m a b
channelProcessor Channel m a b
f =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do let composite :: Composite m (Stream m b)
composite =
           do Signal m a
sa <- forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Composite m (Signal m a)
streamSignal Stream m a
xs
              Signal m b
sb <- forall (m :: * -> *) a b.
Channel m a b -> Signal m a -> Composite m (Signal m b)
runChannel Channel m a b
f Signal m a
sa
              forall (m :: * -> *) a.
MonadDES m =>
Signal m a -> Composite m (Stream m a)
signalStream Signal m b
sb
     (Stream m b
ys, DisposableEvent m
h) <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
EventLift t m =>
Event m a -> t m a
liftEvent forall a b. (a -> b) -> a -> b
$
                forall (m :: * -> *) a.
Composite m a
-> DisposableEvent m -> Event m (a, DisposableEvent m)
runComposite Composite m (Stream m b)
composite forall a. Monoid a => a
mempty
     forall (m :: * -> *). MonadDES m => Event m () -> Process m ()
whenCancellingProcess forall a b. (a -> b) -> a -> b
$
       forall (m :: * -> *). DisposableEvent m -> Event m ()
disposeEvent DisposableEvent m
h
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
ys

-- | Convert the specified processor to a signal transform, i.e. the channel. 
--
-- The processor may return data with delay as the values are requested by demand.
-- Consider using the 'arrivalSignal' function to provide with the information
-- about the time points at which the signal was actually triggered.
--
-- The point is that the 'Stream' used in the 'Processor' is requested outside, 
-- while the 'Signal' used in 'Channel' is triggered inside. They are different by nature.
-- The former is passive, while the latter is active.
--
-- The resulting channel may be a root of space leak as it uses an internal queue to store
-- the values received from the input stream. Consider using 'queuedProcessorChannel' that
-- allows specifying the bounded queue in case of need.
processorChannel :: MonadDES m => Processor m a b -> Channel m a b
{-# INLINABLE processorChannel #-}
processorChannel :: forall (m :: * -> *) a b.
MonadDES m =>
Processor m a b -> Channel m a b
processorChannel (Processor Stream m a -> Stream m b
f) =
  forall (m :: * -> *) a b.
(Signal m a -> Composite m (Signal m b)) -> Channel m a b
Channel forall a b. (a -> b) -> a -> b
$ \Signal m a
sa ->
  do Stream m a
xs <- forall (m :: * -> *) a.
MonadDES m =>
Signal m a -> Composite m (Stream m a)
signalStream Signal m a
sa
     let ys :: Stream m b
ys = Stream m a -> Stream m b
f Stream m a
xs
     forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Composite m (Signal m a)
streamSignal Stream m b
ys

-- | Like 'channelProcessor' but allows specifying an arbitrary queue for storing the signal values,
-- for example, the bounded queue.
queuedChannelProcessor :: MonadDES m
                          => (b -> Event m ())
                          -- ^ enqueue
                          -> Process m b
                          -- ^ dequeue
                          -> Channel m a b
                          -- ^ the channel
                          -> Processor m a b
                          -- ^ the processor
{-# INLINABLE queuedChannelProcessor #-}
queuedChannelProcessor :: forall (m :: * -> *) b a.
MonadDES m =>
(b -> Event m ())
-> Process m b -> Channel m a b -> Processor m a b
queuedChannelProcessor b -> Event m ()
enqueue Process m b
dequeue Channel m a b
f =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do let composite :: Composite m (Stream m b)
composite =
           do Signal m a
sa <- forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Composite m (Signal m a)
streamSignal Stream m a
xs
              Signal m b
sb <- forall (m :: * -> *) a b.
Channel m a b -> Signal m a -> Composite m (Signal m b)
runChannel Channel m a b
f Signal m a
sa
              forall (m :: * -> *) a.
MonadDES m =>
(a -> Event m ())
-> Process m a -> Signal m a -> Composite m (Stream m a)
queuedSignalStream b -> Event m ()
enqueue Process m b
dequeue Signal m b
sb
     (Stream m b
ys, DisposableEvent m
h) <- forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
EventLift t m =>
Event m a -> t m a
liftEvent forall a b. (a -> b) -> a -> b
$
                forall (m :: * -> *) a.
Composite m a
-> DisposableEvent m -> Event m (a, DisposableEvent m)
runComposite Composite m (Stream m b)
composite forall a. Monoid a => a
mempty
     forall (m :: * -> *). MonadDES m => Event m () -> Process m ()
whenCancellingProcess forall a b. (a -> b) -> a -> b
$
       forall (m :: * -> *). DisposableEvent m -> Event m ()
disposeEvent DisposableEvent m
h
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream Stream m b
ys

-- | Like 'processorChannel' but allows specifying an arbitrary queue for storing the signal values,
-- for example, the bounded queue.
queuedProcessorChannel :: MonadDES m =>
                          (a -> Event m ())
                          -- ^ enqueue
                          -> (Process m a)
                          -- ^ dequeue
                          -> Processor m a b
                          -- ^ the processor
                          -> Channel m a b
                          -- ^ the channel
{-# INLINABLE queuedProcessorChannel #-}
queuedProcessorChannel :: forall (m :: * -> *) a b.
MonadDES m =>
(a -> Event m ())
-> Process m a -> Processor m a b -> Channel m a b
queuedProcessorChannel a -> Event m ()
enqueue Process m a
dequeue (Processor Stream m a -> Stream m b
f) =
  forall (m :: * -> *) a b.
(Signal m a -> Composite m (Signal m b)) -> Channel m a b
Channel forall a b. (a -> b) -> a -> b
$ \Signal m a
sa ->
  do Stream m a
xs <- forall (m :: * -> *) a.
MonadDES m =>
(a -> Event m ())
-> Process m a -> Signal m a -> Composite m (Stream m a)
queuedSignalStream a -> Event m ()
enqueue Process m a
dequeue Signal m a
sa
     let ys :: Stream m b
ys = Stream m a -> Stream m b
f Stream m a
xs
     forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Composite m (Signal m a)
streamSignal Stream m b
ys

-- | A processor that adds the information about the time points at which 
-- the original stream items were received by demand.
arrivalProcessor :: MonadDES m => Processor m a (Arrival a)
{-# INLINABLE arrivalProcessor #-}
arrivalProcessor :: forall (m :: * -> *) a. MonadDES m => Processor m a (Arrival a)
arrivalProcessor = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall (m :: * -> *) a.
MonadDES m =>
Stream m a -> Stream m (Arrival a)
arrivalStream

-- | A processor that delays the input stream by one step using the specified initial value.
delayProcessor :: MonadDES m => a -> Processor m a a
{-# INLINABLE delayProcessor #-}
delayProcessor :: forall (m :: * -> *) a. MonadDES m => a -> Processor m a a
delayProcessor a
a0 = forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. MonadDES m => a -> Stream m a -> Stream m a
delayStream a
a0

-- | Removes one level of the computation, projecting its bound processor into the outer level.
joinProcessor :: MonadDES m => Process m (Processor m a b) -> Processor m a b
{-# INLINABLE joinProcessor #-}
joinProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
Process m (Processor m a b) -> Processor m a b
joinProcessor Process m (Processor m a b)
m =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs ->
  forall (m :: * -> *) a. Process m (a, Stream m a) -> Stream m a
Cons forall a b. (a -> b) -> a -> b
$
  do Processor Stream m a -> Stream m b
f <- Process m (Processor m a b)
m
     forall (m :: * -> *) a. Stream m a -> Process m (a, Stream m a)
runStream forall a b. (a -> b) -> a -> b
$ Stream m a -> Stream m b
f Stream m a
xs

-- | Takes the next processor from the list after the current processor fails because of cancelling the underlying process.
failoverProcessor :: MonadDES m => [Processor m a b] -> Processor m a b
{-# INLINABLE failoverProcessor #-}
failoverProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
[Processor m a b] -> Processor m a b
failoverProcessor [Processor m a b]
ps =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream m a
xs -> forall (m :: * -> *) a. MonadDES m => [Stream m a] -> Stream m a
failoverStream [forall (m :: * -> *) a b.
Processor m a b -> Stream m a -> Stream m b
runProcessor Processor m a b
p Stream m a
xs | Processor m a b
p <- [Processor m a b]
ps]

-- | Show the debug messages with the current simulation time.
traceProcessor :: MonadDES m
                  => Maybe String
                  -- ^ the request message
                  -> Maybe String
                  -- ^ the response message
                  -> Processor m a b
                  -- ^ a processor
                  -> Processor m a b
{-# INLINABLE traceProcessor #-}
traceProcessor :: forall (m :: * -> *) a b.
MonadDES m =>
Maybe String -> Maybe String -> Processor m a b -> Processor m a b
traceProcessor Maybe String
request Maybe String
response (Processor Stream m a -> Stream m b
f) =
  forall (m :: * -> *) a b.
(Stream m a -> Stream m b) -> Processor m a b
Processor forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a.
MonadDES m =>
Maybe String -> Maybe String -> Stream m a -> Stream m a
traceStream Maybe String
request Maybe String
response forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream m a -> Stream m b
f