DChan.hs

-- ----------------------------------------------------------------------------

{- |
  Module     : Holumbus.Distribution.DChan
  Copyright  : Copyright (C) 2009 Stefan Schmidt
  License    : MIT

  Maintainer : Stefan Schmidt (stefanschmidt@web.de)
  Stability  : experimental
  Portability: portable
  Version    : 0.1
  
  This module offers a distributed channel datatype (DChan).

  It is similar to Control.Concurrent.Chan, except that you can use it
  between multiple processes on different computers. You can access a
  DChan (reading and writing) from your local process as well as from
  another one.
-}

-- ----------------------------------------------------------------------------

module Holumbus.Distribution.DChan 
(
  -- * datatypes
    DChan
  
  -- * creating and closing channels
  , newDChan
  , newRemoteDChan
  , closeDChan
  
  -- * operations on a channel
  , writeDChan
  , readDChan
  , tryReadDChan
  , tryWaitReadDChan
  , isEmptyDChan
)
where

import           Control.Concurrent.Chan
import           Data.Binary
import qualified Data.ByteString.Lazy as B
import           System.IO
import           System.Log.Logger
import           System.Timeout

import           Holumbus.Distribution.DNode.Base

localLogger :: String
localLogger = "Holumbus.Distribution.DChan"


dChanType :: DResourceType
dChanType = mkDResourceType "DCHAN"

mkDChanEntry :: (Binary a) => DChanReference a -> DResourceEntry
mkDChanEntry d = DResourceEntry {
    dre_Dispatcher   = dispatchDChanRequest d 
  }


data DChanRequestMessage
  = DCMReqRead
  | DCMReqWrite B.ByteString
  | DCMReqIsEmpty
  deriving (Show)

instance Binary DChanRequestMessage where
  put(DCMReqRead)    = putWord8 1
  put(DCMReqWrite a) = putWord8 2 >> put a
  put(DCMReqIsEmpty) = putWord8 3
  get
    = do
      t <- getWord8
      case t of
        1 -> return (DCMReqRead)
        2 -> get >>= \a -> return (DCMReqWrite a)
        3 -> return (DCMReqIsEmpty)
        _ -> error "DChanRequestMessage: wrong encoding"

  
data DChanResponseMessage
  = DCMRspRead B.ByteString
  | DCMRspWrite
  | DCMRspIsEmpty Bool
  deriving (Show)


instance Binary DChanResponseMessage where
  put(DCMRspRead a)    = putWord8 1 >> put a
  put(DCMRspWrite)     = putWord8 2
  put(DCMRspIsEmpty b) = putWord8 3 >> put b
  get
    = do
      t <- getWord8
      case t of
        1 -> get >>= \a -> return (DCMRspRead a)
        2 -> return (DCMRspWrite)
        3 -> get >>= \b -> return (DCMRspIsEmpty b)
        _ -> error "DChanResponseMessage: wrong encoding"


dispatchDChanRequest :: (Binary a) => DChanReference a -> DNodeId -> Handle -> IO () 
dispatchDChanRequest dch _ hdl
  = do
    debugM localLogger "dispatcher: getting message from handle"
    raw <- getByteStringMessage hdl
    let msg = (decode raw)
    -- debugM localLogger $ "dispatcher: Message: " ++ show msg
    case msg of
      (DCMReqRead)    -> handleRead dch hdl
      (DCMReqWrite d) -> handleWrite dch (decode d) hdl
      (DCMReqIsEmpty) -> handleIsEmpty dch hdl

    
-- | The DChan datatype.
--   Notice that this datatype implements the Data.Binary typeclass.
--   That means that you can pass a DChan, so that another computer
--   can access the channel.
data DChan a
  = DChanLocal DResourceAddress (Chan a)
  | DChanRemote DResourceAddress

instance Binary (DChan a) where
  put(DChanLocal a _) = put a
  put(DChanRemote a)  = put a
  get = get >>= \a -> return (DChanRemote a)


data DChanReference a = DChanReference DResourceAddress (Chan a) 


-- | Creates a new DChan on the local computer. The first parameter
--   is the name of the Channel which could be used in other processes to
--   access this stream. If you leave it empty, a random Id will be created.
newDChan :: (Binary a) => String -> IO (DChan a)
newDChan s
  = do
    dra <- genLocalResourceAddress dChanType s
    c <- newChan
    let dch = (DChanLocal dra c)
        dcr = (DChanReference dra c)
        dce = (mkDChanEntry dcr)
    addLocalResource dra dce
    return dch
    
-- TODO merge this with newDChan?
-- | Creates a reference to a DChan which was created in a different
--   process.
--   The first parameter is the name of the resource and the second one
--   the name of the node.
newRemoteDChan :: String -> String -> IO (DChan a)
newRemoteDChan r n
  = do
    return $ DChanRemote dra
    where
    dra = mkDResourceAddress dChanType r n


-- | Closes a DChan object, could not be used anymore after this call.
closeDChan :: (DChan a) -> IO ()
closeDChan (DChanLocal dra _)
  = do
    delLocalResource dra
closeDChan (DChanRemote dra)
  = do
    delForeignResource dra



requestRead :: (Binary a) => Handle -> IO a
requestRead hdl
  = do
    putByteStringMessage (encode $ DCMReqRead) hdl
    raw <- getByteStringMessage hdl
    let rsp = (decode raw)
    case rsp of
      (DCMRspRead d)  -> return $ decode d
      _ -> error "DChan - requestRead: invalid response"
    

handleRead :: (Binary a) => DChanReference a -> Handle -> IO ()
handleRead (DChanReference _ ch) hdl
  = do
    a <- readChan ch
    putByteStringMessage (encode $ DCMRspRead $ encode a) hdl


requestWrite :: (Binary a) => a -> Handle -> IO ()
requestWrite a hdl
  = do 
    putByteStringMessage (encode $ DCMReqWrite $ encode a) hdl
    raw <- getByteStringMessage hdl
    let rsp = (decode raw)
    case rsp of
      (DCMRspWrite) -> return ()
      _ -> error "DChan - requestWrite: invalid response"
    

handleWrite :: DChanReference a -> a -> Handle -> IO ()
handleWrite (DChanReference _ ch) a hdl
  = do
    writeChan ch a
    putByteStringMessage (encode $ DCMRspWrite) hdl


requestIsEmpty :: Handle -> IO Bool
requestIsEmpty hdl
  = do
    putByteStringMessage (encode $ DCMReqIsEmpty) hdl
    raw <- getByteStringMessage hdl
    let rsp = (decode raw)
    case rsp of
      (DCMRspIsEmpty b) -> return b
      _ -> error "DChan - requestIsEmpty: invalid response"
    

handleIsEmpty :: DChanReference a -> Handle -> IO ()
handleIsEmpty (DChanReference _ ch) hdl
  = do
    b <- isEmptyChan ch
    putByteStringMessage (encode $ DCMRspIsEmpty b) hdl 


-- | Writes data to a DChan.
writeDChan :: (Binary a) => DChan a -> a -> IO ()
writeDChan (DChanLocal _ c) v
  = do
    writeChan c v
writeDChan (DChanRemote a) v
  = do
    unsafeAccessForeignResource a (requestWrite v)


-- | Reads data from a DChan, blocks if DChan is empty.
readDChan :: (Binary a) => DChan a -> IO a
readDChan (DChanLocal _ c)
  = do
    readChan c
readDChan (DChanRemote a)
  = do
    unsafeAccessForeignResource a requestRead


-- | Tries to read data from a DChan, if the DChan is empty,
--   the function return with Nothing.
tryReadDChan :: (Binary a) => DChan a -> IO (Maybe a)
tryReadDChan dc
  = do
    empty <- isEmptyDChan dc
    if (not empty) 
      then do timeout 1000 (readDChan dc)
      else do return Nothing


-- | Reads data from a DChan. If the channel is empty, it waits
--   for a given time (in microseconds) an returns immediately
--   when new data arrives, otherwise it return Nothing.
tryWaitReadDChan :: (Binary a) => DChan a -> Int -> IO (Maybe a) 
tryWaitReadDChan dc t = timeout t (readDChan dc)
    

-- | Tests, if a DChan is empty.
isEmptyDChan :: DChan a -> IO Bool
isEmptyDChan (DChanLocal _ c)
  = do
    isEmptyChan c
isEmptyDChan (DChanRemote a)
  = do
    unsafeAccessForeignResource a requestIsEmpty