-- ----------------------------------------------------------------------------
{- |
Module : Holumbus.Network.Port
Copyright : Copyright (C) 2008 Stefan Schmidt
License : MIT
Maintainer : Stefan Schmidt (stefanschmidt@web.de)
Stability : experimental
Portability: portable
Version : 0.1
Stream and Port datatype for internal an external process communication.
Useful for communikation of distributed systems.
-}
-- ----------------------------------------------------------------------------
{-# OPTIONS_GHC -fno-warn-unused-binds #-} -- for unused record field selectors
module Holumbus.Network.Port
(
-- * Constants
time1
, time10
, time30
, time60
, time120
, timeIndefinitely
-- * Datatypes
, SocketId(..) -- reexport from core
, MessageType
, Message
, StreamName
, StreamType(..)
, Stream
, Port
-- * Message-Operations
, getMessageType
, getMessageData
, getGenericData
-- * Global-Operations
, setPortRegistry
-- * Stream-Operations
, newGlobalStream
, newLocalStream
, newPrivateStream
, newStream
, closeStream
, isEmptyStream
, readStream
, readStreamMsg
, tryReadStream
, tryReadStreamMsg
, tryWaitReadStream
, tryWaitReadStreamMsg
, withStream
-- * Port-Operations
, newPortFromStream
, newPort
, newGlobalPort
, isPortLocal
, send
, sendWithGeneric
, sendWithMaybeGeneric
, writePortToFile
, readPortFromFile
-- * Debug
, printStreamController
)
where
import Control.Concurrent
import Control.Exception
import Control.Monad
import Data.Binary
import qualified Data.ByteString.Lazy as B
import Data.Char
import qualified Data.Map as Map
import Data.Maybe
import qualified Data.Set as Set
import Data.Time
import Network
import System.IO
import System.IO.Unsafe
import System.Log.Logger
import System.Timeout
import Text.XML.HXT.Arrow
import Holumbus.Common.Utils ( handleAll )
import Holumbus.Network.Site
import Holumbus.Network.Core
import Holumbus.Network.PortRegistry
import qualified Holumbus.Data.MultiMap as MMap
localLogger :: String
localLogger = "Holumbus.Network.Port"
-- -----------------------------------------------------------------------------
-- Constants
-- -----------------------------------------------------------------------------
-- | One second
time1 :: Int
time1 = 1000000
-- | 10 seconds
time10 :: Int
time10 = 10000000
-- | 30 seconds
time30 :: Int
time30 = 30000000
-- | 60 seconds
time60 :: Int
time60 = 60000000
-- | 120 seconds
time120 :: Int
time120 = 120000000
-- | Wait how long it takes
timeIndefinitely :: Int
timeIndefinitely = -1
-- | The default starting port number
defaultPort :: PortNumber
defaultPort = 9000
-- | The default maximal port number
maxPort :: PortNumber
maxPort = 40000
-- ----------------------------------------------------------------------------
-- Message-Datatype
-- ----------------------------------------------------------------------------
-- | Message Type
-- Is it an internal Message or does it come from an external Node?
data MessageType = MTInternal | MTExternal
deriving (Show)
instance Binary MessageType where
put MTInternal = putWord8 1
put MTExternal = putWord8 2
get
= do
t <- getWord8
case t of
1 -> return MTInternal
_ -> return MTExternal
-- | Message Datatype.
-- We are sending additional information, to do debugging
data (Show a, Binary a) => Message a = Message {
msg_Type :: ! MessageType -- ^ the message-type
, msg_Receiver :: ! StreamName -- ^ the name of the destination stream
, msg_Data :: ! a -- ^ the data
, msg_Generic :: ! (Maybe B.ByteString) -- ^ some generic data -- could be another port
, msg_ReceiverSocket :: ! (Maybe SocketId) -- ^ socket to which the message is send (DEBUG)
, msg_SenderSocket :: ! (Maybe SocketId) -- ^ socket from which the message was send (DEBUG)
, msg_Send_time :: ! UTCTime -- ^ timestamp from the sender (DEBUG)
, msg_Receive_time :: ! UTCTime -- ^ timestamp from the receiver (DEBUG)
} deriving (Show)
instance (Show a, Binary a) => Binary (Message a) where
put (Message t r d g rs ss t1 t2)
= do
put t
put r
put d
put g
put rs
put ss
put $ show t1
put $ show t2
get
= do
t <- get
r <- get
d <- get
g <- get
rs <- get
ss <- get
t1Str <- get
t2Str <- get
return $ (Message t r d g rs ss (read t1Str) (read t2Str))
-- ----------------------------------------------------------------------------
-- Port-Datatype
-- ----------------------------------------------------------------------------
-- | The port datatype.
data Port a = Port { p_StreamName :: StreamName -- ^ the name of the destination stream
, p_SocketId :: Maybe SocketId
}
deriving (Show)
instance (Show a, Binary a) => Binary (Port a) where
put (Port sn soid) = put sn >> put soid
get = do
sn <- get
soid <- get
return (Port sn soid)
instance (Show a, Binary a) => XmlPickler (Port a) where
xpickle = xpPort
xpPort :: PU (Port a)
xpPort =
xpElem "port" $
xpWrap(\(sn, soid) -> Port sn soid, \(Port sn soid) -> (sn, soid)) $
xpPair (xpAttr "name" xpText) (xpOption $ xpickle)
-- ----------------------------------------------------------------------------
-- Stream-Datatype
-- ----------------------------------------------------------------------------
-- | The name of a stream.
type StreamName = String
-- | The stream type, determines the accessibility of a stream
data StreamType = STGlobal | STLocal | STPrivate
deriving (Show, Eq, Ord)
-- | The stream datatype
data Stream a
= Stream {
s_StreamName :: StreamName
, s_SocketId :: SocketId
, s_Type :: StreamType
, s_Channel :: BinaryChannel
}
instance (Show a, Binary a) => Show (Stream a) where
show (Stream sn soid st _)
= "(Stream" ++
" - Name: " ++ show sn ++
" - SocketId: " ++ show soid ++
" - Type: " ++ show st ++
" )"
-- ----------------------------------------------------------------------------
-- StreamController-Datatype
-- ----------------------------------------------------------------------------
-- | A chan datatype for binary messages
type BinaryChannel = (Chan (Message B.ByteString))
-- | The stream controller datatype.
-- We need this to keep a log of all used streams in the program and to
-- use multiple streams per unix-socket. The access information for the
-- PortRegistry is also placed here. There should only be one stream
-- controller per program. To ensure this, we use a global reference to this
-- data object.
data StreamControllerData = StreamControllerData
(Maybe SocketId)
(Maybe GenericRegistry)
Int
(Map.Map StreamName (BinaryChannel, PortNumber, StreamType))
(Map.Map PortNumber (ThreadId, HostName))
(MMap.MultiMap PortNumber StreamName)
instance Show StreamControllerData where
show (StreamControllerData ds _ si stm sem pom)
= "StreamControllerData:\n"
++ " DefaultSocket:\t" ++ show ds
++ " lastStreamId:\t" ++ show si
++ " Streams:\t" ++ show (Map.keys stm)
++ " ServerMap:\t" ++ show sem
++ " PortMap:\t" ++ show pom
-- | The stream controller.
type StreamController = MVar StreamControllerData
-- ----------------------------------------------------------------------------
-- StreamController-Operations
-- ----------------------------------------------------------------------------
-- | A direct access to the program stream controller.
{-# NOINLINE streamController #-}
streamController :: StreamController
streamController
= do
unsafePerformIO $ newMVar emptyStreamControllerData
where
emptyStreamControllerData
= StreamControllerData
Nothing
Nothing
0
Map.empty
Map.empty
MMap.empty
-- | Sets the link to the PortRegistry in the stream controller
setPortRegistry :: (PortRegistry r) => r -> IO ()
setPortRegistry r
= modifyMVar streamController $
\(StreamControllerData s _ i sm pm pmm) ->
return ((StreamControllerData s (Just $ mkGenericRegistry r) i sm pm pmm),())
-- | Gets the next generated and program unique stream name.
getNextStreamName :: IO (StreamName)
getNextStreamName
= modifyMVar streamController $
\(StreamControllerData s r i sm pm pmm) ->
do
let i' = i + 1
let n = "$" ++ show i'
let scd' = StreamControllerData s r i' sm pm pmm
return (scd',n)
-- | Test, if a user defined stream name is valid.
-- A stream name may only contain numbers and characters.
isValidStreamName :: StreamName -> Bool
isValidStreamName [] = False
isValidStreamName sn = null $ filter isForbiddenChar sn
where
isForbiddenChar c = not $ isAlphaNum c
-- | Test, if the stream name is valid and if it is not already used in the
-- program.
validateStreamName :: Maybe StreamName -> IO (StreamName)
validateStreamName (Nothing) = getNextStreamName
validateStreamName (Just sn)
| isValidStreamName sn
= do
taken <- withMVar streamController $
\(StreamControllerData _ _ _ sm _ _) ->
return $ Map.member sn sm
if (taken)
then do error "stream name already exists"
else do return sn
| otherwise
= error "invalid stream name"
-- | Opens a unix-socket on the given port number. If no port number is
-- specified, the default port number will be used. It is not problem that
-- two ports share the same unix-socket. The stream controller will handle
-- the incomming messages to the right stream.
openSocket :: Maybe PortNumber -> IO (SocketId)
-- get/start the default socket
openSocket (Nothing)
= modifyMVar streamController $
\scd@(StreamControllerData s r i sm pm pmm) ->
do
case s of
(Nothing) ->
do
res <- startStreamServer defaultPort maxPort
case res of
(Nothing) ->
error "Port: getDefaultPort: unable to open defaultsocket"
(Just (soid@(SocketId hn pn), tid)) ->
do
let pm' = Map.insert pn (tid,hn) pm
let s' = Just soid
let scd' = StreamControllerData s' r i sm pm' pmm
return (scd',soid)
(Just soid) ->
return (scd,soid)
-- get/start the new socket
openSocket (Just pn)
= modifyMVar streamController $
\scd@(StreamControllerData s r i sm pm pmm) ->
do
let mp = Map.lookup pn pm
case mp of
(Nothing) ->
do
res <- startStreamServer pn pn
case res of
(Nothing) ->
error "Port: getDefaultPort: unable to open socket"
(Just (soid@(SocketId hn _), tid)) ->
do
let pm' = Map.insert pn (tid,hn) pm
let scd' = StreamControllerData s r i sm pm' pmm
return (scd',soid)
(Just (_,hn)) ->
return (scd,SocketId hn pn)
-- | Closes a socket, only possible, if the unix-socket was used by only one
-- stream.
closeSocket :: SocketId -> IO ()
closeSocket soid@(SocketId _ pn)
= modifyMVar streamController $
\scd@(StreamControllerData s r i sm pm pmm) ->
do
if (isNothing s || soid == fromJust s)
-- don't delete the defaultSocket
then do
return (scd,())
else do
if (MMap.member pn pmm)
-- if the port is still used by other streams, keep it
then do
return (scd,())
-- close the socket and delete it from the controller
else do
let (tid, _) = fromJust $ Map.lookup pn pm
stopStreamServer tid
let pm' = Map.delete pn pm
let scd' = StreamControllerData s r i sm pm' pmm
return (scd',())
-- | Registers a stream at the stream controller.
registerStream :: Stream a -> IO ()
registerStream st
= modifyMVar streamController $
\(StreamControllerData s r i sm pm pmm) ->
do
let sm' = Map.insert sn (ch,pn,ty) sm
let pmm' = MMap.insert pn sn pmm
return ((StreamControllerData s r i sm' pm pmm'), ())
where
(SocketId _ pn) = s_SocketId st
ch = s_Channel st
sn = s_StreamName st
ty = s_Type st
-- | Deletes a stream from the stream controller.
unregisterStream :: Stream a -> IO ()
unregisterStream st
= modifyMVar streamController $
\(StreamControllerData s r i sm pm pmm) ->
do
let sm' = Map.delete sn sm
let pmm' = MMap.deleteElem pn sn pmm
return ((StreamControllerData s r i sm' pm pmm'), ())
where
(SocketId _ pn) = s_SocketId st
sn = s_StreamName st
-- | Registers a stream at the global PortRegistry.
registerGlobalPort :: Stream a -> IO ()
registerGlobalPort (Stream sn soid STGlobal _)
= do
r <- withMVar streamController $
\(StreamControllerData _ r' _ _ _ _) -> return r'
case r of
(Just r') -> registerPort sn soid r'
(Nothing) -> errorM localLogger $ "registerGlobalPort: no portregistry while handling port " ++ sn
registerGlobalPort _ = return ()
-- | Deletes a stream from the global PortRegistry.
unregisterGlobalPort :: Stream a -> IO ()
unregisterGlobalPort (Stream sn _ STGlobal _)
= do
r <- withMVar streamController $
\(StreamControllerData _ r' _ _ _ _) -> return r'
case r of
(Just r') -> unregisterPort sn r'
(Nothing) -> errorM localLogger $ "unregisterGlobalPort: no portregistry while handling port " ++ sn
unregisterGlobalPort _ = return ()
-- | Get the unix-socket of a stream to create a global port.
getGlobalPort :: StreamName -> IO (Maybe SocketId)
getGlobalPort sn
= do
r <- withMVar streamController $
\(StreamControllerData _ r' _ _ _ _) -> return r'
case r of
(Just rp) ->
do
handleAll (\e -> do
errorM localLogger $ "getGlobalPort: error while getting port: " ++ sn ++ " exception: " ++ show e
return Nothing
) $ do
lookupPort sn rp
(Nothing) -> do
errorM localLogger $ "getGlobalPort: no portregistry found while getting port: " ++ sn
return Nothing
-- | Gets the names of all stream from a specified port number.
getStreamNamesForPort :: PortNumber -> IO (Set.Set StreamName)
getStreamNamesForPort pn
= withMVar streamController $
\(StreamControllerData _ _ _ _ _ pmm) ->
return $ MMap.lookup pn pmm
-- | Gets the data of a stream from the stream controller.
getStreamData :: StreamName -> IO (Maybe (BinaryChannel, PortNumber, StreamType))
getStreamData sn
= withMVar streamController $
\(StreamControllerData _ _ _ sm _ _) ->
return $ Map.lookup sn sm
-- -----------------------------------------------------------------------------
-- Message-Operations
-- -----------------------------------------------------------------------------
-- | Encodes a message into a bytestring.
encodeMessage :: (Show a, Binary a) => Message a -> Message B.ByteString
encodeMessage (Message t n d g r s t1 t2) = (Message t n (encode d) g r s t1 t2)
-- | Decodes a message from a bytestring.
decodeMessage :: (Show a, Binary a) => Message B.ByteString -> Message a
decodeMessage (Message t n d g r s t1 t2) = (Message t n (decode d) g r s t1 t2)
-- | Creates a new message with the current time.
newMessage :: (Show a, Binary a) => MessageType -> StreamName -> a -> Maybe B.ByteString -> IO (Message a)
newMessage t n d g
= do
time <- getCurrentTime
return (Message t n d g Nothing Nothing time time)
-- | Gets the type of a message.
getMessageType :: (Show a, Binary a) => Message a -> MessageType
getMessageType = msg_Type
-- | Gets the data of a message.
getMessageData :: (Show a, Binary a) => Message a -> a
getMessageData = msg_Data
-- | Gets the generic data (usually the return port) of a message.
getGenericData :: (Show a, Binary a) => Message a -> (Maybe B.ByteString)
getGenericData = msg_Generic
-- | Gets the distination stream name of the message.
getMessageReceiver :: (Show a, Binary a) => Message a -> StreamName
getMessageReceiver = msg_Receiver
-- | Sets the receive time of a message.
updateReceiveTime :: (Show a, Binary a) => Message a -> IO (Message a)
updateReceiveTime msg
= do
time <- getCurrentTime
return (msg {msg_Receive_time = time})
-- | Sets the receiver unix-socket of the message.
updateReceiverSocket :: (Show a, Binary a) => Message a -> SocketId -> Message a
updateReceiverSocket msg soId = msg { msg_ReceiverSocket = Just soId }
-- | Sets the sender unix-socket of the message.
updateSenderSocket :: (Show a, Binary a) => Message a -> SocketId -> Message a
updateSenderSocket msg soId = msg { msg_SenderSocket = Just soId }
-- ----------------------------------------------------------------------------
-- Stream-Operations
-- ----------------------------------------------------------------------------
-- | Creates a new global stream.
newGlobalStream :: (Show a, Binary a) => StreamName -> IO (Stream a)
newGlobalStream n = newStream STGlobal (Just n) Nothing
-- | Creates a new local stream.
newLocalStream :: (Show a, Binary a) => Maybe StreamName -> IO (Stream a)
newLocalStream sn = newStream STLocal sn Nothing
-- | Creates a new private stream.
newPrivateStream :: (Show a, Binary a) => Maybe StreamName -> IO (Stream a)
newPrivateStream sn = newStream STPrivate sn Nothing
-- | General function for creating a new stream.
newStream
:: (Show a, Binary a)
=> StreamType -> Maybe StreamName -> Maybe PortNumber
-> IO (Stream a)
newStream STGlobal Nothing _ = error "newStream: global ports always need a name"
newStream st n pn
= do
ch <- newChan
sn <- validateStreamName n
infoM localLogger $ "opening socket for " ++ sn
soid <- openSocket pn
let s = Stream sn soid st ch
infoM localLogger $ "registering " ++ show sn ++ " at controller"
registerStream s
infoM localLogger $ "registering " ++ show sn ++ " at registry"
registerGlobalPort s
return s
-- | Closes a stream.
closeStream :: (Show a, Binary a) => Stream a -> IO ()
closeStream s
= do
infoM localLogger $ "unregistering " ++ show sn ++ " at registry"
unregisterGlobalPort s
infoM localLogger $ "unregistering " ++ show sn ++ " at controller"
unregisterStream s
infoM localLogger $ "closing socket for " ++ sn
closeSocket (s_SocketId s)
where
sn = s_StreamName s
-- | Writes a message to the channel of the stream.
-- This function is not public, so you have to write to a stream throug its
-- port
writeChannel :: Chan (Message B.ByteString) -> Message B.ByteString -> IO ()
writeChannel ch msg
= do
newMsg <- updateReceiveTime msg
writeChan ch newMsg
-- | Test, if the stream contains new messages.
isEmptyStream :: Stream a -> IO Bool
isEmptyStream s
= do
isEmptyChan (s_Channel s)
-- | Reads the data packet of the next message from a stream.
-- If stream is empty, this function will block until a new message arrives.
readStream :: (Show a, Binary a) => Stream a -> IO a
readStream s
= do
msg <- readStreamMsg s
return (msg_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.
readStreamMsg :: (Show a, Binary a) => Stream a -> IO (Message a)
readStreamMsg s
= do
debugM localLogger "PORT: readStreamMsg 1"
res <- readChan (s_Channel s)
debugM localLogger "PORT: readStreamMsg 2"
return $ decodeMessage res
--TODO is there a better way for non-blocking?
-- | Helper function for doing a non blocking stream action.
tryStreamAction :: Stream a -> (Stream a -> IO (b)) -> IO (Maybe b)
tryStreamAction s f
= do
empty <- isEmptyStream s
if (not empty)
then do timeout 1000 (f s)
else do return Nothing
-- | Helper function for doing a timed stream action.
tryWaitStreamAction :: Stream a -> (Stream a -> IO (b)) -> Int -> IO (Maybe b)
tryWaitStreamAction s f t
= do
debugM localLogger "tryWaitStreamAction: waiting..."
r <- timeout t (f s)
debugM localLogger "tryWaitStreamAction: ...finished"
let debugResult = maybe "nothing received" (\_ -> "value found") r
debugM localLogger $ "tryWaitStreamAction: " ++ debugResult
return r
-- | Reads the data packet of the next message from a stream.
-- If stream is empty, this function will immediately return with Nothing.
tryReadStream :: (Show a, Binary a) => Stream a -> IO (Maybe a)
tryReadStream s
= do
tryStreamAction s readStream
-- | Reads the next message from a stream (data packet + message header).
-- If stream is empty, this function will immediately return with Nothing.
tryReadStreamMsg :: (Show a, Binary a) => Stream a -> IO (Maybe (Message a))
tryReadStreamMsg s
= do
tryStreamAction s readStreamMsg
-- | 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.
tryWaitReadStream :: (Show a, Binary a) => Stream a -> Int -> IO (Maybe a)
tryWaitReadStream s t
= do
tryWaitStreamAction s readStream t
-- | 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.
tryWaitReadStreamMsg :: (Show a, Binary a) => Stream a -> Int -> IO (Maybe (Message a))
tryWaitReadStreamMsg s t
= do
tryWaitStreamAction s readStreamMsg t
-- | Encapsulates a stream.
-- A new stream is created, then some user-action is done an after that the
-- stream is closed.
withStream :: (Show a, Binary a) => (Stream a -> IO b) -> IO b
withStream f
= do
debugM localLogger "withStream: creating new stream"
s <- newStream STLocal Nothing Nothing
debugM localLogger "withStream: new stream created"
res <- f s
closeStream s
return res
-- ----------------------------------------------------------------------------
-- StreamServer-Operations
-- ----------------------------------------------------------------------------
-- | Starts a new thread which will listen on a unix-socket for new messages
-- and delegate them to their streams.
startStreamServer :: PortNumber -> PortNumber -> IO (Maybe (SocketId, ThreadId))
startStreamServer actPo maxPo
= do
res <- startSocket (streamDispatcher) actPo maxPo
case res of
Nothing ->
return Nothing
(Just (tid, hn, po)) ->
return (Just (SocketId hn po, tid))
-- | Stops a thread listening to a unix-socket.
stopStreamServer :: ThreadId -> IO ()
stopStreamServer sId
= do
me <- myThreadId
debugM localLogger $ "stopping server... with threadId: " ++ show sId ++ " - form threadId: " ++ show me
{- 6.8 throwDynTo sId me -}
throwTo sId (ThreadIdException me)
yield
-- | Delegates new incomming messages on a unix-socket to their streams.
streamDispatcher :: SocketId -> Handle -> HostName -> PortNumber -> IO ()
streamDispatcher (SocketId _ ownPo) hdl hn po
= do
debugM localLogger "streamDispatcher: getting message from handle"
raw <- getMessage hdl
let msg = (decode raw)::Message B.ByteString
debugM localLogger $ "streamDispatcher: Message: " ++ show msg
let sn = getMessageReceiver msg
sns <- getStreamNamesForPort ownPo
sd <- getStreamData sn
if (Set.member sn sns)
-- if the socket knows the stream
then do
case sd of
(Just (_,_,STPrivate)) ->
warningM localLogger $ "streamDispatcher: received msg for private stream " ++ sn
(Just (ch,_,_)) ->
do
debugM localLogger "streamDispatcher: writing message to channel"
writeChannel ch $ updateSenderSocket msg (SocketId hn po)
debugM localLogger "streamDispatcher: message written to channel"
(Nothing) ->
do
errorM localLogger $ "streamDispatcher: no channel for stream " ++ sn
-- if the stream is unknown, log this as error
else do
warningM localLogger $ "streamDispatcher: received msg for unknown stream " ++ sn
-- -----------------------------------------------------------------------------
-- Port Operations
-- -----------------------------------------------------------------------------
-- | Creates a new Port, which is bound to a stream.
newPortFromStream :: Stream a -> IO (Port a)
newPortFromStream s
= do
let sn = s_StreamName s
let soid = s_SocketId s
return $ Port sn (Just soid)
-- | Creates a new port from a streamname and its socketId.
newPort :: StreamName -> Maybe SocketId -> IO (Port a)
newPort sn soid = return $ Port sn soid
-- | Creates a new port to a global stream, only its name is needed.
newGlobalPort :: StreamName -> IO (Port a)
newGlobalPort sn = return $ Port sn Nothing
-- | Test, if a port is local.
isPortLocal :: Port a -> IO Bool
isPortLocal (Port sn mbSoid)
= do
sd <- getStreamData sn
case sd of
(Just (_,po,_)) ->
do
case mbSoid of
(Just s1) ->
do
sid <- getSiteId
let hn = getSiteHost sid
let s2 = SocketId hn po
return (s1 == s2)
(Nothing) ->
do
return False
(Nothing) ->
return False
-- | Send data to the stream of the port.
-- The data is send via network, if the stream is located on an external
-- processor
send :: (Show a, Binary a) => Port 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 :: (Show a, Binary a) => Port a -> a -> B.ByteString -> IO ()
sendWithGeneric p d rp = sendWithMaybeGeneric p d (Just rp)
-- | Like "sendWithGeneric", but the generic data is optional
sendWithMaybeGeneric :: (Show a, Binary a) => Port a -> a -> Maybe B.ByteString -> IO ()
sendWithMaybeGeneric p@(Port sn mbsoid) d rp
= do
local <- isPortLocal p
if local
-- we have a local stream
then do
msg <- newMessage MTInternal sn d rp
sd <- getStreamData sn
case sd of
(Just (ch,_,_)) ->
do
writeChannel ch $ encodeMessage msg
_ ->
do
errorM localLogger $ "sendWithMaybeGeneric: no channel found for stream " ++ sn
else do
-- we have an external stream
case mbsoid of
-- we know it's port
(Just so@(SocketId hn po)) ->
do
msg <- newMessage MTExternal sn d rp
let raw = encode $ encodeMessage $ updateReceiverSocket msg so
sendRequest (putMessage raw) hn (PortNumber po)
return ()
-- we don't know it's port
(Nothing) ->
do
extsoid <- getGlobalPort sn
case extsoid of
(Just so@(SocketId hn po)) ->
do
msg <- newMessage MTExternal sn d rp
let raw = encode $ encodeMessage $ updateReceiverSocket msg so
sendRequest (putMessage raw) hn (PortNumber po)
return ()
(Nothing) ->
do
errorM localLogger errorMsg
error errorMsg
where
errorMsg = "sendWithMaybeGeneric: global port not found for stream " ++ sn
-- | Writes a port-description to a file.
-- Quite useful fpr sharing ports between programs
writePortToFile :: (Show a, Binary a) => Port a -> FilePath -> IO ()
writePortToFile p fp
= do
bracket
(openFile fp WriteMode)
(hClose)
(\hdl ->
do
enc <- return (encode $ p)
hPutStrLn hdl ((show $ B.length enc) ++ " ")
B.hPut hdl enc
)
-- | Reads a port-description from a file.
-- Quite useful fpr sharing ports between programs
readPortFromFile :: (Show a, Binary a) => FilePath -> IO (Port a)
readPortFromFile fp
= do
bracket
(openFile fp ReadMode)
(hClose)
(\hdl ->
do
pkg <- liftM words $ hGetLine hdl
raw <- B.hGet hdl (read $ head pkg)
p <- return (decode raw)
return p
)
-- ----------------------------------------------------------------------------
-- Debugging
-- ----------------------------------------------------------------------------
-- | Prints the internal data of the stream controller to stdout,
-- useful for debugging.
printStreamController :: IO ()
printStreamController
= withMVar streamController $
\scd ->
do
putStrLn "StreamController:"
putStrLn $ show scd