content: -- ----------------------------------------------------------------------------
  
  {- |
    Module     : Holumbus.Distribution.DStreamPort
    Copyright  : Copyright (C) 2009 Stefan Schmidt
    License    : MIT
  
    Maintainer : Stefan Schmidt (stefanschmidt@web.de)
    Stability  : experimental
    Portability: portable
    Version    : 0.1
  
    This module offers distributed streams and ports.
  
    Because a DChan allows external read access, the idea came up to
    split a DChan into two parts: a stream and a port. A stream only allows
    you to read data from it. The read-access is limited to the local process which
    created the stream. To send data to a stream, you need a port. This can be used
    on forgein nodes to send data to your local stream.
    
  -}
  
  -- ----------------------------------------------------------------------------
  
  module Holumbus.Distribution.DStreamPort
  (
    -- * datatypes
      DStream
    , DPort
    , StreamPortMessage(..)
    
    -- * creating and closing a stream
    , newDStream
    , closeDStream
  
    -- * operations on a stream
    , isEmptyDStream  
    , receive
    , receiveMsg
    , tryReceive
    , tryReceiveMsg
    , tryWaitReceive
    , tryWaitReceiveMsg
    , withStream
      
    -- * creating a port
    , newDPortFromStream
    , newDPort
  
    -- * operations on a port
    , send
    , sendWithGeneric
    , sendWithMaybeGeneric
  )
  where
  
  import           Data.Binary
  --import           Holumbus.Common.MRBinary
  import qualified Data.ByteString.Lazy as B
  import           System.Log.Logger
  
  import           Holumbus.Distribution.DChan
  
  
  localLogger :: String
  localLogger = "Holumbus.Distribution.DStreamPort"
  
  -- ----------------------------------------------------------------------------
  -- | Message Datatype.
  --   We are sending additional information, to do debugging
  data (Binary a) => StreamPortMessage a = StreamPortMessage {
      spm_Data           :: ! a                    -- ^ the data  
    , spm_Generic        :: ! (Maybe B.ByteString) -- ^ some generic data -- could be another port
    }
  
  instance (Binary a) => Binary (StreamPortMessage a) where
    put(StreamPortMessage d g) = put d >> put g
    get = get >>= \d -> get >>= \g -> return $ (StreamPortMessage d g)
  
  
  -- ----------------------------------------------------------------------------
  newtype DStream a = DStream (DChan (StreamPortMessage a))
  
  
  -- | Creates a new local stream.
  newDStream :: (Binary a) => String -> IO (DStream a) 
  newDStream s
    = do
      dc <- newDChan s
      return (DStream dc)
  
  
  -- | Closes a stream.
  closeDStream :: DStream a -> IO ()
  closeDStream (DStream dc) = closeDChan dc
  
  
  -- | Tests, if a stream has no more data to read.
  isEmptyDStream :: DStream a -> IO Bool
  isEmptyDStream (DStream dc) = isEmptyDChan dc
  
  
  getMaybeMsgData :: (Binary a) => Maybe (StreamPortMessage a) -> Maybe a
  getMaybeMsgData (Nothing)  = Nothing
  getMaybeMsgData (Just msg) = Just (spm_Data msg)
  
  -- | Reads the data packet of the next message from a stream.
  --   If stream is empty, this function will block until a new message arrives.
  receive :: (Binary a) => DStream a -> IO a
  receive s = (receiveMsg s) >>= \msg -> return (spm_Data msg)
  
  
  -- | Reads the next message from a stream (data packet + message header).
  --   If stream is empty, this function will block until a new message arrives.
  receiveMsg :: (Binary a) => DStream a -> IO (StreamPortMessage a)
  receiveMsg (DStream dc) = readDChan dc
  
  
  -- | Reads the data packet of the next message from a stream.
  --   If stream is empty, this function will immediately return with Nothing.
  tryReceive :: (Binary a) => DStream a -> IO (Maybe a)
  tryReceive s = tryReceiveMsg s >>= \msg -> return $ getMaybeMsgData msg
  
  
  -- | Reads the next message from a stream (data packet + message header).
  --   If stream is empty, this function will immediately return with Nothing.
  tryReceiveMsg :: (Binary a) => DStream a -> IO (Maybe (StreamPortMessage a))
  tryReceiveMsg (DStream dc) = tryReadDChan dc
  
  
  -- | Reads the data packet of the next message from a stream.
  --   If stream is empty, this function will wait for new messages until the 
  --   time is up and if no message has arrived, return with Nothing.
  tryWaitReceive :: (Binary a) => DStream a -> Int -> IO (Maybe a)
  tryWaitReceive s t = (tryWaitReceiveMsg s t) >>= \msg -> return $ getMaybeMsgData msg
  
  
  -- | Reads the next message from a stream (data packet + message header).
  --   If stream is empty, this function will wait for new messages until the 
  --   time is up and if no message has arrived, return with Nothing.
  tryWaitReceiveMsg :: (Binary a) => DStream a -> Int -> IO (Maybe (StreamPortMessage a))
  tryWaitReceiveMsg (DStream dc) t = tryWaitReadDChan dc t
  
      
  -- | Encapsulates a stream.
  --   A new stream is created, then some user-action is done an after that the
  --   stream is closed. 
  withStream :: (Binary a) => (DStream a -> IO b) -> IO b
  withStream f
    = do
      debugM localLogger "withStream: creating new stream"
      s <- newDStream ""
      debugM localLogger "withStream: new stream created"
      res <- f s
      closeDStream s
      return res
  
     
  -- ----------------------------------------------------------------------------
  newtype DPort a = DPort (DChan (StreamPortMessage a))
  
  instance (Binary a) => Binary (DPort a) where
    put(DPort dc) = put dc
    get = get >>= \dc -> return $ (DPort dc)
      
      
  -- | Creates a new Port, which is bound to a stream.
  newDPortFromStream :: DStream a -> IO (DPort a)
  newDPortFromStream (DStream dc) = return (DPort dc)
  
  
  -- | Creates a new port from a streamname and its socketId.
  --   The first parameter is the name of the resource and the second one
  --   the name of the node.
  newDPort :: String -> String -> IO (DPort a)
  newDPort r n = newRemoteDChan r n >>= \dc -> return (DPort dc)
      
     
  -- | Send data to the stream of the port.
  --   The data is send via network, if the stream is located on an external
  --   processor
  send :: (Binary a) => DPort a -> a -> IO ()
  send p d = sendWithMaybeGeneric p d Nothing
  
  
  -- | Like "send", but here we can give some generic data (e.g. a port for reply 
  --   messages).
  sendWithGeneric :: (Binary a) => DPort a -> a -> B.ByteString -> IO ()
  sendWithGeneric p d rp = sendWithMaybeGeneric p d (Just rp) 
  
  
  -- | Like "sendWithGeneric", but the generic data is optional
  sendWithMaybeGeneric :: (Binary a) => DPort a -> a -> Maybe B.ByteString -> IO ()
  sendWithMaybeGeneric (DPort dc) d rp = writeDChan dc (StreamPortMessage d rp)
  