> {-# LANGUAGE GeneralizedNewtypeDeriving, MultiParamTypeClasses, TypeFamilies, TypeOperators #-}

> module Control.Concurrent.Actors (
>
>     {- | 
>     Here we demonstrate a binary tree of actors that supports concurrent
>     insert and query operations:
>      
>     > import Control.Concurrent.Actors
>     > import Control.Applicative
>     > import Control.Concurrent.MVar
>     > 
>     > -- the actor equivalent of a Nil leaf node:
>     > nil :: Behavior Operation
>     > nil = Receive $ do
>     >     (Query _ var) <- received 
>     >     send var False -- signal Int is not present in tree
>     >     return nil     -- await next message
>     > 
>     >    <|> do          -- else, Insert received
>     >     l <- spawn nil -- spawn child nodes
>     >     r <- spawn nil
>     >     branch l r . val <$> received  -- create branch from inserted val
>     >     
>     > -- a branch node with a value 'v' and two children
>     > branch :: Node -> Node -> Int -> Behavior Operation    
>     > branch l r v = loop where
>     >     loop = Receive $ do
>     >         m <- received 
>     >         case compare (val m) v of
>     >              LT -> send l m
>     >              GT -> send r m
>     >              EQ -> case m of -- signal Int present in tree:
>     >                         (Query _ var) -> send var True
>     >                         _             -> return ()
>     >         return loop
>     > 
>     > type Node = Mailbox Operation
>     > 
>     > -- operations supported by the network:
>     > data Operation = Insert { val :: Int }
>     >                | Query { val :: Int
>     >                        , sigVar :: Mailbox Bool }
>     > 
>     > insert :: Node -> Int -> IO ()
>     > insert t = send t . Insert
>     > 
>     > query :: Node -> Int -> IO Bool
>     > query t a = do
>     >     -- turn an MVar into a Mailbox actors can send to with 'out'
>     >     v <- newEmptyMVar
>     >     send t (Query a $ out v)
>     >     takeMVar v
>     
>     You can use the tree defined above in GHCi:
>     
>     >>> :l TreeExample.hs 
>     Ok
>     >>> t <- spawn nil
>     >>> query t 7
>     False
>     >>> insert t 7
>     >>> query t 7
>     True
>
>     -}
>
>     -- * Actor Behaviors
>       Action()
>     , Behavior(..)
>     -- ** Composing and Transforming Behaviors
>     , (<.|>)
>
>     -- * Available actions
>     -- ** Message passing
>     , Mailbox()
>     , out
>     , send , send' , (<->)
>     , received
>     , guardReceived
>     -- ** Spawning actors
>     , Sources(), Joined
>   --, (:-:)(..)
>     , spawn
>     -- *** Mailboxes and scoping
>     {- | 
>     Straightforward use of the 'spawn' function will be sufficient for
>     forking actors in most cases, but launching mutually-communicating actors
>     presents a problem.
>      
>     In cases where a 'Behavior' needs access to its own 'Mailbox' or that of 
>     an actor that must be forked later, the 'MonadFix' instance should be
>     used. GHC\'s \"Recursive Do\" notation make this especially easy:
>      
>     > {-# LANGUAGE DoRec #-}
>     > beh = Receive $ do
>     >     i <- received
>     >     -- similar to the scoping in a "let" block:
>     >     rec b1 <- spawn (senderTo b2)
>     >         b2 <- spawn (senderTo b1)
>     >         b3 <- spawn (senderTo b3)
>     >     -- send initial messages to actors spawned above:
>     >     send b3 i
>     >     send b2 "first"
>     >     yield
>
>     -}
>
>     -- ** Building an actor computation
>     {- | 
>     An actor computation can be halted immediately by calling 'yield',
>     a synonym for 'mzero'. When an 'Action' calling @yield@ is composed with
>     another using @\<|\>@ the second takes over processing the /same/ input
>     which the former @yield@-ed on.
>
>     Here is an example of a computation using 'guard' which returns @mzero@ if
>     the test is false:
>
>     > foo c n = Receive $ 
>     >       do i <- received
>     >          guard (n<10)
>     >          send c i
>     >          return (foo c $ n+1)
>     >
>     >   <|> do i <- received -- same as the 'i' above
>     >          send c $ "TENTH INPUT: "++i
>     >          return (foo c 0)
>
>     The @Monoid@ instance for 'Behavior' works on the same principle.
>     -}
>     , yield
>     , receive
>
>     -- ** Composing and Transforming Mailboxes
>     {- |
>     We offer some operations to split and combine 'Mailbox'es of sum and
>     product types. 
>     -}
>     , coproductMb
>     , contraProduct
>     , zipMb
>     , contraFanin
>     , contraFanout
>
>     -- * Utility functions
>     {- | 
>     These are useful for debugging 'Behavior's
>     -}
>     , runBehavior_
>     , runBehavior 
>
>     -- * Useful predefined @Behavior@s
>     , printB
>     , putStrB
>     , signalB
>     , constB
>
>     ) where
>
> import Control.Monad
> import Control.Applicative
> import Control.Monad.Reader(ask)
> import qualified Data.Foldable as F
> import Control.Monad.IO.Class
> import Control.Concurrent(forkIO)
> import Data.Monoid
> import Control.Arrow((***),(&&&),(|||))
>
> -- from the contravariant package 
> import Data.Functor.Contravariant
> -- from the chan-split package
> import Control.Concurrent.Chan.Split
> -- internal:
> import Control.Concurrent.Actors.Behavior

> type Sender a = Op (IO ()) a
>
> mailbox :: (a -> IO ()) -> Mailbox a
> mailbox = Mailbox . Op
>
> runMailbox :: Mailbox a -> a -> IO ()
> runMailbox = getOp . sender
>
> mkMessages :: OutChan a -> Messages a
> mkMessages = Messages . readChan
>
> -- | One can 'send' a messages to a @Mailbox@ where it will be processed
> -- according to an actor\'s defined 'Behavior'
> --
> -- > type Joined (Mailbox a) = a
> newtype Mailbox a = Mailbox { sender :: Sender a }
>       deriving (Contravariant)

> -- | Convert the input side of a @SplitChan@ to a @Mailbox@. Useful for 
> -- sending data out from an actor system via a channel created in IO.
> out :: (SplitChan i x)=> i a -> Mailbox a
> out = mailbox . writeChan

> newtype Messages a = Messages { readMsg :: IO a }
>       deriving (Functor) 
>
> -- Not sure how to derive this or if possible:
> instance SplitChan Mailbox Messages where
>     readChan = readMsg
>     writeChan = runMailbox
>
> instance NewSplitChan Mailbox Messages where
>     newSplitChan = (out *** mkMessages) `fmap` newSplitChan

> coproductMb :: Mailbox a -> Mailbox b -> Mailbox (Either a b) 
> coproductMb m1 m2 = mailbox $ either (writeChan m1) (writeChan m2)
>
> zipMb :: Mailbox a -> Mailbox b -> Mailbox (a,b) 
> zipMb m1 m2 = mailbox $ \(a,b) -> writeChan m1 a >> writeChan m2 b
>

> -- | > contraProduct = contramap Left &&& contramap Right
> contraProduct :: Contravariant f => f (Either a b) -> (f a, f b)
> contraProduct = contramap Left &&& contramap Right
>
> -- | > contraFanin f g = contramap (f ||| g)
> contraFanin :: Contravariant f => (b -> a) -> (c -> a) -> f a -> f (Either b c)
> contraFanin f g = contramap (f ||| g)
>
> -- | > contraFanout f g = contramap (f &&& g)
> contraFanout :: Contravariant f=> (a -> b) -> (a -> c) -> f (b,c) -> f a
> contraFanout f g = contramap (f &&& g)

> infixl 3 <.|>
>
> -- | Sequence two @Behavior@s. After the first 'yield's the second takes over,
> -- discarding the message the former was processing. See also the 'Monoid'
> -- instance for @Behavior@.
> -- 
> -- > b <.|> b' = b `mappend` constB b'
> (<.|>) :: Behavior i -> Behavior i -> Behavior i
> b <.|> b' = b `mappend` constB b'

> -- | Immediately give up processing an input, perhaps relinquishing the input
> -- to an 'Alternative' computation or exiting the actor.
> -- 
> -- > yield = mzero
> yield :: Action i a
> yield = mzero
>
> -- | Useful to make defining a continuing Behavior more readable as a
> -- \"receive block\", e.g.
> --
> -- > pairUpAndSendTo mb = Receive $ do
> -- >     a <- received
> -- >     receive $ do
> -- >         b <- received
> -- >         send mb (b,a)
> -- >         return (pairUpAndSendTo mb)
> --
> -- Defined as: 
> --
> -- > receive = return . Receive
> receive :: Action i (Behavior i) -> Action i (Behavior i)
> receive = return . Receive

> -- | Return the message received to start this 'Action' block. /N.B/ the value
> -- returned here does not change between calls in the same 'Action'.
> --
> -- > received = ask
> received :: Action i i
> received = ask

> -- | Return 'received' message matching predicate, otherwise 'yield'.
> --
> -- > guardReceived p = ask >>= \i-> guard (p i) >> return i
> guardReceived :: (i -> Bool) -> Action i i
> guardReceived p = ask >>= \i-> guard (p i) >> return i

> -- | Send a message asynchronously to an actor receiving from Mailbox. See
> -- also 'out' for converting other types of chans to 'Mailbox'.
> --  
> -- > send b = liftIO . writeChan b
> send :: (MonadIO m)=> Mailbox a -> a -> m ()
> send b = liftIO . writeChan b

> -- | A strict 'send':
> --
> -- > send' b a = a `seq` send b a
> send' :: (MonadIO m)=> Mailbox a -> a -> m ()
> send' b a = a `seq` send b a

> infixr 1 <->
>
> -- | Like 'send' but supports chaining sends by returning the Mailbox.
> -- Convenient for initializing an Actor with its first input after spawning,
> -- e.g.
> --
> -- >     do mb <- 0 <-> spawn foo
> (<->) :: (MonadIO m)=> a -> m (Mailbox a) -> m (Mailbox a)
> a <-> mmb = mmb >>= \mb-> send mb a >> return mb

> -- | We extend the actor model to support joining (or synchronizing) multiple
> -- 'Mailbox'es to a single 'Behavior' input type, using a new class with an
> -- associated type. Functionality is best explained by example:
> --
> -- Spawn an actor returning it's 'Mailbox', and send it its first message:
> -- 
> -- > sumTuple :: Behavior (Int, Int)
> -- >
> -- > do b <- spawn sumTuple
> -- >    send b (4, 1) 
> -- >    ...
> --
> -- But now we would like our @sumTuple@ actor to receive each number from a different 
> -- concurrent actor:
> --
> -- > do (b1, b2) <- spawn sumTuple
> -- >    b3 <- spawn (multipliesBy2AndSendsTo b1)
> -- >    send b3 2
> -- >    send b2 1
> -- >    ...
> --
> -- Lastly spawn an actor that starts immediately on an infinite supply of @()@s,
> -- and supplies an endless stream of @Int@s to @sumTuple@
> --
> -- > do (b1, b2) <- spawn sumTuple
> -- >    () <- spawn (sendsIntsTo b2)
> -- >    send b1 4
> -- >    ...
> class Sources s where
>     type Joined s
>     newJoinedChan :: IO (s, Messages (Joined s)) -- private

> -- | Fork an actor performing the specified 'Behavior'. /N.B./ an actor
> -- begins execution of its 'headBehavior' only after a message becomes
> -- available to process; for sending an initial message to an actor right
> -- after 'spawn'ing it, ('<|>') can be convenient.
> spawn :: (MonadIO m, Sources s)=> Behavior (Joined s) -> m s
> spawn b = liftIO $ do
>     (srcs, msgs) <- newJoinedChan
>     let runner b' = readChan msgs >>= runBehaviorStep b' >>= F.mapM_ runner
>     void $ forkIO (runner b)
>     return srcs

> instance Sources (Mailbox a) where
>     type Joined (Mailbox a) = a
>     newJoinedChan = newSplitChan

> instance (Sources a, Sources b)=> Sources (a,b) where
>     type Joined (a,b) = (Joined a, Joined b)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         let m' = Messages $ liftM2 (,) (readMsg ma) (readMsg mb)
>         return ((sa,sb), m')

> instance (Sources a, Sources b, Sources c, Sources d, Sources e, Sources f, Sources g)=> Sources (a,b,c,d,e,f,g) where
>     type Joined (a,b,c,d,e,f,g) = (Joined a, Joined b,Joined c,Joined d,Joined e,Joined f,Joined g)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         (sc, mc) <- newJoinedChan
>         (sd, md) <- newJoinedChan
>         (se, me) <- newJoinedChan
>         (sf, mf) <- newJoinedChan
>         (sg, mg) <- newJoinedChan
>         let m' = Messages $ (,,,,,,) <$> readMsg ma <*> readMsg mb <*> readMsg mc <*> readMsg md <*> readMsg me <*> readMsg mf <*> readMsg mg 
>         return ((sa,sb,sc,sd,se,sf,sg), m')
>
> instance (Sources a, Sources b, Sources c, Sources d, Sources e, Sources f)=> Sources (a,b,c,d,e,f) where
>     type Joined (a,b,c,d,e,f) = (Joined a, Joined b,Joined c,Joined d,Joined e,Joined f)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         (sc, mc) <- newJoinedChan
>         (sd, md) <- newJoinedChan
>         (se, me) <- newJoinedChan
>         (sf, mf) <- newJoinedChan
>         let m' = Messages $ (,,,,,) <$> readMsg ma <*> readMsg mb <*> readMsg mc <*> readMsg md <*> readMsg me <*> readMsg mf
>         return ((sa,sb,sc,sd,se,sf), m')
>
> instance (Sources a, Sources b, Sources c, Sources d, Sources e)=> Sources (a,b,c,d,e) where
>     type Joined (a,b,c,d,e) = (Joined a, Joined b,Joined c,Joined d,Joined e)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         (sc, mc) <- newJoinedChan
>         (sd, md) <- newJoinedChan
>         (se, me) <- newJoinedChan
>         let m' = Messages $ (,,,,) <$> readMsg ma <*> readMsg mb <*> readMsg mc <*> readMsg md <*> readMsg me
>         return ((sa,sb,sc,sd,se), m')
>
> instance (Sources a, Sources b, Sources c, Sources d)=> Sources (a,b,c,d) where
>     type Joined (a,b,c,d) = (Joined a, Joined b,Joined c,Joined d)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         (sc, mc) <- newJoinedChan
>         (sd, md) <- newJoinedChan
>         let m' = Messages $ (,,,) <$> readMsg ma <*> readMsg mb <*> readMsg mc <*> readMsg md 
>         return ((sa,sb,sc,sd), m')
>
> instance (Sources a, Sources b, Sources c)=> Sources (a,b,c) where
>     type Joined (a,b,c) = (Joined a, Joined b,Joined c)
>     newJoinedChan = do
>         (sa, ma) <- newJoinedChan
>         (sb, mb) <- newJoinedChan
>         (sc, mc) <- newJoinedChan
>         let m' = Messages $ (,,) <$> readMsg ma <*> readMsg mb <*> readMsg mc 
>         return ((sa,sb,sc), m')

> -- | > type Joined () = ()
> --
> -- Represents an endless supply of @()@s. Allows 'spawn'-ing
> -- a @Behavior ()@ that starts immediately and loops until it 'yield'-s, e.g.
> -- 
> -- > do () <- spawn startsImmediately -- :: Behavior ()
> instance Sources () where
>     type Joined () = ()
>     newJoinedChan = 
>         return ((), Messages $ return ())

> {-# RULES "spawn_" spawn = spawn_  #-}
> spawn_ :: (MonadIO m)=> Behavior () -> m ()
> spawn_ = liftIO . void . forkIO . runBehavior_

> -- | Run a @Behavior ()@ in the main thread, returning when the computation
> -- exits.
> runBehavior_ :: Behavior () -> IO ()
> runBehavior_ b = runBehavior b $ repeat ()
>
> -- | run a 'Behavior' in the IO monad, taking its \"messages\" from the list.
> runBehavior :: Behavior a -> [a] -> IO ()
> runBehavior b (a:as) = runBehaviorStep b a >>= F.mapM_ (`runBehavior` as)
> runBehavior _ _      = return ()

> -- | Prints all messages to STDOUT in the order they are received,
> -- 'yield'-ing /immediately/ after @n@ inputs are printed.
> printB :: (Show s, Eq n, Num n)=> n -> Behavior s
> printB = contramap (unlines . return . show) . putStrB

> -- | Like 'printB' but using @putStr@.
> putStrB :: (Eq n, Num n)=> n -> Behavior String
> putStrB 0 = mempty --special case when called directly w/ 0
> putStrB n = Receive $ do
>     s <- received
>     liftIO $ putStr s
>     guard (n /= 1)
>     return $ putStrB (n-1)

> -- | Sends a @()@ to the passed chan. This is useful with 'mappend' for
> -- signalling the end of some other 'Behavior'.
> --
> -- > signalB c = Receive (send c () >> yield)
> signalB :: Mailbox () -> Behavior i
> signalB c = Receive (send c () >> yield)

> -- | A @Behavior@ that discard its first input, returning the passed Behavior
> -- for processing subsequent inputs. Useful with 'Alternative' or 'Monoid'
> -- compositions when one wants to ignore the leftover 'yield'ed message.
> --
> -- > constB = Receive . return
> constB :: Behavior i -> Behavior i
> constB = Receive . return