packages feed

Holumbus-Distribution (empty) → 0.0.1

raw patch · 34 files changed

+5485/−0 lines, 34 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, containers, editline, haskell98, hslogger, hxt, network, time, unix

Files

+ Examples/ClientServer/Client.hs view
@@ -0,0 +1,61 @@++module Main(main) where++import           Holumbus.Common.Debug+import           Holumbus.Common.Logging+import           Holumbus.Common.Utils+import           Holumbus.Network.PortRegistry.PortRegistryPort+import           Holumbus.Network.Communication++version :: String+version = "Holumbus-Communication Client 0.1"+++main :: IO ()+main+  = do+    putStrLn version+    handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+      do+      initializeLogging []+      p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+      setPortRegistry p+      client <- initializeData+      waitForInput client+      deinitializeData client++waitForInput :: Client -> IO ()+waitForInput client+  = do+    line <- getLine+    case line of+      "exit"  -> +        return ()+      "debug" -> +        do+        printDebug client+        waitForInput client+      _ -> +        waitForInput client+++dispatch +  :: String +  -> IO (Maybe String)+dispatch msg+  = do+    putStrLn $ "message arrived: " ++ msg+    return Nothing+++initializeData :: IO (Client)+initializeData +  = do+    client <- newClient "server" Nothing dispatch+    return client+++deinitializeData :: Client -> IO ()+deinitializeData client+  = do+    closeClient client
+ Examples/ClientServer/Makefile view
@@ -0,0 +1,31 @@+# Making FileSystem-Standalone Programs++HOME		= ../..++SOURCE		= .++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= Client Server++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+
+ Examples/ClientServer/Server.hs view
@@ -0,0 +1,75 @@++module Main(main) where++import           Holumbus.Common.Debug+import           Holumbus.Common.Logging+import           Holumbus.Common.Utils+import           Holumbus.Network.PortRegistry.PortRegistryPort+import           Holumbus.Network.Communication+++version :: String+version = "Holumbus-Communication Server 0.1"+++main :: IO ()+main+  = do+    putStrLn version+    handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+      do+      initializeLogging []+      p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+      setPortRegistry p+      server <- initializeData+      waitForInput server+      deinitializeData server+++waitForInput :: Server -> IO ()+waitForInput server+  = do+    line <- getLine+    case line of+      "exit"  -> +        return ()+      "debug" -> +        do+        printDebug server+        waitForInput server+      _ -> +        waitForInput server+++dispatch +  :: String +  -> IO (Maybe String)+dispatch msg+  = do+    putStrLn $ "message arrived: " ++ msg+    return Nothing+++register :: IdType -> ClientPort -> IO ()+register i cp+  = do+    putStrLn $ "register request: Id: " ++ show i ++ " -- Port: " ++ show cp+++unregister :: IdType -> ClientPort -> IO ()+unregister i cp+  = do+    putStrLn $ "unregister request: Id: " ++ show i ++ " -- Port: " ++ show cp+++initializeData :: IO (Server)+initializeData +  = do+    server <- newServer "server" Nothing dispatch (Just register) (Just unregister)+    return server+++deinitializeData :: Server -> IO ()+deinitializeData server+  = do+    closeServer server
+ Examples/Makefile view
@@ -0,0 +1,13 @@+# Making all examples++ALLEXAMPLES       = ClientServer Ports+++all :+	$(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i all ;)++wc :+	@wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+	$(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i $@ ;)
+ Examples/Ports/Makefile view
@@ -0,0 +1,31 @@+# Making FileSystem-Standalone Programs++HOME		= ../..++SOURCE		= .++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= Single Sender Receiver++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) p.port port.xml+
+ Examples/Ports/Receiver.hs view
@@ -0,0 +1,98 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : +  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import           Control.Concurrent+import qualified Data.ByteString.Lazy as B++import           Holumbus.Common.Logging+import           Holumbus.Common.Utils+import           Holumbus.Network.Port+import           Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Console.Console as Console+++type StringStream = Stream String+type StringPort   = Port String+++newStringStream :: IO StringStream+newStringStream = newLocalStream Nothing+++decodeStringPort :: Maybe B.ByteString -> Maybe StringPort+decodeStringPort = decodeMaybe+++version :: String+version = "Ports-Demo Receiver 0.1"+++main :: IO ()+main+  = do+    initializeLogging []+    putStrLn version    +    putStrLn "Begin"+    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"+    setPortRegistry reg+    printStreamController+    putStrLn "initialising streams..."+    s <- newStringStream+    p <- newPortFromStream s+    globalS <- newStream  STGlobal  (Just "global")  (Just 10002)+    putStrLn "streams initialised"+    writePortToFile p "p.port"+    forkIO $ printMessages "global" globalS+    Console.handleUserInput createConsole ()+++printMessages :: String -> StringStream -> IO ()+printMessages info s+  = do+    msg <- readStreamMsg s+    putStrLn $ "port: " ++ info+    putStrLn "Incoming Message:"+    putStrLn $ show msg+    let replyPort = decodeStringPort $ getGenericData msg+    let text      = getMessageData msg+    case replyPort of+        (Nothing) ->  +          do+          putStrLn "no reply port in message"+        (Just p) ->+          do+          send p ("ECHO:" ++ text)+    printMessages info s+    +    +createConsole :: Console.ConsoleData ()+createConsole =+  Console.addConsoleCommand "debug" printDebug "prints internal state of the filesystem (DEBUG)" $ +  Console.addConsoleCommand "version" printVersion "prints the version" $ +  Console.initializeConsole++ +printDebug :: () -> [String] -> IO ()+printDebug _ _+  = do+    printStreamController+  +printVersion :: () -> [String] -> IO ()+printVersion _ _+  = do+    putStrLn version+
+ Examples/Ports/Sender.hs view
@@ -0,0 +1,105 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : +  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where+    +import           Data.List+-- import           Data.Binary++import           Holumbus.Common.Logging+import           Holumbus.Network.Port+import           Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Console.Console as Console+++version :: String+version = "Ports-Demo Sender 0.1"+++main :: IO ()+main+  = do+    initializeLogging []+    putStrLn version+    putStrLn "Begin"+    printStreamController +    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"+    setPortRegistry reg+    p <- readPortFromFile "p.port"+    Console.handleUserInput createConsole p++        +createConsole :: Console.ConsoleData (Port String)+createConsole =+  -- Console.addConsoleCommand "send" sendMsg "sends a message" $+  -- Console.addConsoleCommand "sendP" sendPMsg "sends a message to the private port of the receiver" $+  -- Console.addConsoleCommand "sendN" sendNMsg "sends a message to the named port of the receiver" $+  Console.addConsoleCommand "sendG" sendGMsg "sends a message to the global port of the receiver" $+  Console.addConsoleCommand "debug" printDebug "prints internal state of the filesystem (DEBUG)" $ +  Console.addConsoleCommand "version" printVersion "prints the version" $ +  Console.initializeConsole++{-+sendMsg :: StringPort -> [String] -> IO ()+sendMsg p line+  = do+    respStream <- newStringStream+    respPort <- newPortFromStream respStream+    let msg = intercalate " " line+    sendWithGeneric p msg (encode respPort)+    response <- tryWaitReadStreamMsg respStream time10+    case response of+      -- if no response+      Nothing -> putStrLn "no response in 10 seconds"          +      -- handle the response+      (Just r) -> putStrLn $ "ResponseMessage:\n" ++ show r+    closeStream respStream+    +sendPMsg :: StringPort -> [String] -> IO ()+sendPMsg _ line+  = do+    privPort <- (newPort "private" "localhost" 10000)::IO StringPort+    let msg = intercalate " " line+    send privPort msg+ + +sendNMsg :: StringPort -> [String] -> IO ()+sendNMsg _ line+  = do+    namedPort <- (newPort "named" "localhost" 10001)::IO StringPort+    let msg = intercalate " " line+    send namedPort msg+-}++sendGMsg :: (Port String) -> [String] -> IO ()+sendGMsg _ line+  = do+    globalPort <- (newGlobalPort "global")::IO (Port String)+    let msg = intercalate " " line+    send globalPort msg+++printDebug :: (Port String) -> [String] -> IO ()+printDebug _ _+  = do+    printStreamController+++++printVersion :: (Port String) -> [String] -> IO ()+printVersion _ _+  = do+    putStrLn version
+ Examples/Ports/Single.hs view
@@ -0,0 +1,108 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : +  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import           Data.Binary+import qualified Data.ByteString.Lazy as B++import           Holumbus.Common.Logging+import           Holumbus.Network.Port+++version :: String+version = "Holumbus-Ports Single-Demo 0.1"+++type StringStream = Stream String+type StringPort   = Port String+++decodeMaybe :: (Binary a) => Maybe B.ByteString -> Maybe a+decodeMaybe Nothing = Nothing+decodeMaybe (Just b) = (Just $ decode b)+++decodeStringPort :: Maybe B.ByteString -> Maybe StringPort+decodeStringPort = decodeMaybe++++main :: IO ()+main+  = do+    initializeLogging []+    putStrLn version+    +    putStrLn ""+    putStrLn "opening private Streams and Ports"+    +    printStreamController+    stream1 <- (newLocalStream (Just "stream1")):: IO StringStream+    stream2 <- (newLocalStream (Just "stream2")):: IO StringStream+    printStreamController+    +    testCase stream1 stream2 "Hello World"+    +    putStrLn ""+    putStrLn "closing streams..."+    closeStream stream1+    closeStream stream2+    printStreamController+    +    +    +testCase :: StringStream -> StringStream -> String -> IO ()+testCase stream1 stream2 t+  = do+    putStrLn ""+    putStrLn "Begin"++    port1   <- newPortFromStream stream1+    port2   <- newPortFromStream stream2+    +    putStrLn ""+    putStrLn $ "Sending: " ++ t+    sendWithGeneric port1 t (encode port2)+    printStreamController++    putStrLn ""+    putStrLn "Reading result..."+    msg1 <- readStreamMsg stream1+    putStrLn "Result:"+    putStrLn $ "Text: " ++ getMessageData msg1+    putStrLn $ "Msg: " ++ show msg1+    +    putStrLn ""+    putStrLn "echoing result..."+    let replyPort = decodeStringPort $ getGenericData msg1+    let text      = getMessageData msg1+    case replyPort of    +        (Nothing) ->+          do+          putStrLn "no reply port in message"+        (Just p) ->+          do+          send p ("ECHO:" ++ text)+    ++    putStrLn "Reading echo..."+    msg2 <- readStreamMsg stream2+    putStrLn "Result:"+    putStrLn $ "Text: " ++ getMessageData msg2+    putStrLn $ "Msg: " ++ show msg2+        +    putStrLn ""+    putStrLn "End"
+ Holumbus-Distribution.cabal view
@@ -0,0 +1,74 @@+name:          Holumbus-Distribution+version:       0.0.1+license:       OtherLicense+license-file:  LICENSE+author:        Stefan Schmidt, Uwe Schmidt+copyright:     Copyright (c) 2008 Stefan Schmidt, Uwe Schmidt+maintainer:    Stefan Schmidt <sts@holumbus.org>+stability:     experimental+category:      Distributed Computing, Network+synopsis:      intra- and inter-program communication+homepage:      http://holumbus.fh-wedel.de+description:   Holumbus-Distribution offers Erlang-Style mailboxes for an easy implementation+               of distributed systems in Haskell. The mailboxes can be used to exchange messages+               between threads inside a single program or between programs over the network.+build-type:    Simple+cabal-version: >=1.6++extra-source-files:+    README+    Makefile+    Programs/Makefile+    Programs/PortRegistry/Makefile+    Programs/PortRegistry/PortRegistry.hs+    Examples/Makefile+    Examples/ClientServer/Client.hs+    Examples/ClientServer/Makefile+    Examples/ClientServer/Server.hs+    Examples/Ports/Makefile+    Examples/Ports/Receiver.hs+    Examples/Ports/Sender.hs+    Examples/Ports/Single.hs++library+  build-depends:  base       >= 4+                , binary     >= 0.4+                , bytestring >= 0.9+                , containers >= 0.1+                , editline   >= 0.2+                , haskell98  >= 1+                , hslogger   >= 1.0+                , hxt        >= 8.2+                , network    >= 2.1+                , time       >= 1.1+                , unix       >= 2.3++  hs-source-dirs: +                  source+  exposed-modules: Holumbus.Data.AccuMap+                 , Holumbus.Data.KeyMap+                 , Holumbus.Data.MultiMap+                 , Holumbus.Network.Communication+                 , Holumbus.Network.Core+                 , Holumbus.Network.Messages+                 , Holumbus.Network.Port+                 , Holumbus.Network.Site+                 , Holumbus.Network.PortRegistry+                 , Holumbus.Network.PortRegistry.Messages+                 , Holumbus.Network.PortRegistry.PortRegistryData+                 , Holumbus.Network.PortRegistry.PortRegistryPort+                 , Holumbus.Console.Console+                 , Holumbus.Common.Debug+                 , Holumbus.Common.FileHandling+                 , Holumbus.Common.Logging+                 , Holumbus.Common.Threading+                 , Holumbus.Common.Utils++  ghc-options: -Wall -threaded+  extensions: Arrows DeriveDataTypeable ExistentialQuantification++executable PortRegistry+  main-is: PortRegistry.hs+  hs-source-dirs: Programs/PortRegistry source+  ghc-options: -Wall -threaded+  extensions: Arrows DeriveDataTypeable ExistentialQuantification
+ LICENSE view
@@ -0,0 +1,21 @@+The MIT License++Copyright (c) 2008 Stefan Schmidt++Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation the+rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +sell copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE.
+ Makefile view
@@ -0,0 +1,56 @@+# Making all+#+# Building the library and generating documentation is done by cabal,+# only the tests and the examples will be built using make.++# TEST_BASE	= test+EXAMPLES_BASE	= Examples+PROGRAMS_BASE	= Programs++all : configure build doc dist++configure :+	@runhaskell Setup.hs configure++doc	: configure+	@runhaskell Setup.hs haddock --hyperlink-source --hscolour-css=hscolour.css++build	: configure+	@runhaskell Setup.hs build++install :+	@runhaskell Setup.hs install --global++dist	:+	@runhaskell Setup.hs sdist++prof	:+	@runhaskell Setup.hs configure -p+	@runhaskell Setup.hs build++# alltests :+#	$(MAKE) -C $(TEST_BASE) all++examples :+	$(MAKE) -C $(EXAMPLES_BASE) all++programs :	+	$(MAKE) -C $(PROGRAMS_BASE) all++wc :+	@wc -l `find ./source/Holumbus -wholename './_darcs/*' -prune -o -name "*.hs" -print`   +	# $(MAKE) -C $(TEST_BASE) wc+	$(MAKE) -C $(EXAMPLES_BASE) wc+	$(MAKE) -C $(PROGRAMS_BASE) wc++zip:	clean+	@zip -r holumbus_mapreduce$$(date +%Y%m%d_%H%M).zip *++clean :+	@runhaskell Setup.hs clean+	@rm -rf holumbus_mapreduce*.zip+	# $(MAKE) -C $(TEST_BASE) clean+	$(MAKE) -C $(EXAMPLES_BASE) clean+	$(MAKE) -C $(PROGRAMS_BASE) clean++.PHONY	: configure build doc dist clean zip wc examples programs
+ Programs/Makefile view
@@ -0,0 +1,18 @@+# Making all examples++# COMMUNICATION     = Communication/ClientServer Communication/Ports+PORTREGISTRY      = PortRegistry+ALLPROGRAMS       = $(PORTREGISTRY)+++all :+	$(foreach i,$(ALLPROGRAMS),$(MAKE) -C $i all ;)++registry :+	$(foreach i,$(PORTREGISTRY),$(MAKE) -C $i all ;)++wc :+	@wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+	$(foreach i,$(ALLPROGRAMS),$(MAKE) -C $i $@ ;)
+ Programs/PortRegistry/Makefile view
@@ -0,0 +1,30 @@++HOME		= ../..++SOURCE		= .++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= PortRegistry++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) registry.xml registry.port+
+ Programs/PortRegistry/PortRegistry.hs view
@@ -0,0 +1,95 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : +  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import           Holumbus.Common.Logging+import           Holumbus.Common.FileHandling+import           Holumbus.Common.Utils+import           Holumbus.Network.Port+import           Holumbus.Network.PortRegistry+import           Holumbus.Network.PortRegistry.PortRegistryData+import qualified Holumbus.Console.Console as Console+++version :: String+version = "Holumbus-PortRegistry 0.1"+++main :: IO ()+main+  = do+    initializeLogging []+    let sn = "portregistry"+    let po = (Just 9000)+    r <- newPortRegistryData sn po+    p <- getPortRegistryRequestPort r+    writePortToFile p "/tmp/registry.port"+    saveToXmlFile "/tmp/registry.xml" p+    Console.handleUserInput createConsole r +    putStrLn version++    +createConsole :: Console.ConsoleData PortRegistryData+createConsole =+  Console.addConsoleCommand "register" 	 hdlRegister   "registers a port manually" $+  Console.addConsoleCommand "unregister" hdlUnregister "unregisters a port manually" $+  Console.addConsoleCommand "lookup" 	 hdlLookup     "gets the socket id for a port" $+  Console.addConsoleCommand "ports" 	 hdlGetPorts   "lists all ports" $ +  Console.addConsoleCommand "version"    hdlVersion    "prints the version" $+  Console.initializeConsole+++hdlRegister :: PortRegistryData -> [String] -> IO ()+hdlRegister prd opts+  = do+    handleAll (\_ -> putStrLn "usage: register <streamname> <hostname> <socketId>") $+      do+      let sn = head opts+      let hn = head $ tail opts+      let po = fromInteger $ read $ head $ tail $ tail opts+      registerPort sn (SocketId hn po) prd+++hdlUnregister :: PortRegistryData -> [String] -> IO ()+hdlUnregister prd opts+  = handleAll (\_ -> putStrLn "usage: unregister <streamname>") $+      do+      let sn = head opts+      unregisterPort sn prd+++hdlLookup :: PortRegistryData -> [String] -> IO ()+hdlLookup prd opts+  = handleAll (\_ -> putStrLn "usage: lookup <streamname>") $+      do+      let sn = head opts+      soid <- lookupPort sn prd+      putStrLn $ show soid+++hdlGetPorts :: PortRegistryData -> [String] -> IO ()+hdlGetPorts prd _+  = do+    ls <- getPorts prd+    putStrLn $ show ls+++hdlVersion :: PortRegistryData -> [String] -> IO ()+hdlVersion _ _+  = do+    putStrLn version++-- ------------------------------------------------------------
+ README view
@@ -0,0 +1,156 @@+This is the Holumbus-Distribution Library++Version 0.0.1++Stefan Schmidt sts@holumbus.org++http://holumbus.fh-wedel.de+++About+-----++Holumbus is a set of Haskell libraries. This package contains the +Holumbus-Distribution library for building and running a distributed systems.+One of the core elements of this library is an Erlang-like mailbox +system for interchanging messages between different threads or applications. ++This library itself is independent from other Holumbus libraries.+++Documentation:+------------------------------------------++The "Chan" data type form Control.Concurrent.Chan is pretty useful.+You can easily use it as a queue for the consumer-producer-problem. One+thread writes into the channel, the other reads the data from it. But this+only works, if the two threads are running in the same address space.+That means the Chan data type cannot be used for the communication of+distinct applications, but this scenario is a normal use case.++Therefore it would be very useful to have a single framework for inter- +and intra-process communication. One thread can create a mailbox and other +threads can send messages to it, though it should make no difference, +if the threads are in the same address space or not.++This libary offers a "Stream-Port" implementation, borrowed from the+multi-paradigm programming language Mozart/Oz.++A stream in this context is like a mailbox or a receiver. Other threads can send+messages to it via a port. A port in this context has nothing to do with the+port number of a unix-socket. Think of it as a sender.++To address a stream over the network easily, you can give him a unique name.+Then you can create a port, give him that stream name and the messages will+be send directly to the stream. This requires you to start the PortRegistry.++The PortRegistry keeps a log for all global communication streams in the+network. It MUST be started BEFORE all other programs. In future versions this+explicit execution sequence might not be necessary any more, but for now the+FIRST thing you have to start is the PortRegistry. There is ONLY ONE one+instance allowed in the whole communication network (this might also change).++You can find the main parts for the communication in the module+"Holumbus.Network.Port"++This library also contains some parts which might be useful, e.g. some+specialized maps (Holumbus.Data) and a commandline user interface+(Holumbus.Console). They are used in other Holumbus libraries.+++Contents+--------++Examples  Some example applications+Programs  The applications you need to run a distributed system.+source    Source code of the Holumbus-Distribution library.+++Requirements+------------++So far, this library is only tested under Linux, please tell me, if you have +problems under Windows or other OS.+The Holumbus-Distribution library requires at least GHC 6.10 and the +following packages (available via Hackage).++  containers+  hslogger+  network +  unix+  time+  bytestring+  binary+  hxt+++Installation+------------++A Cabal file is provided, therefore Holumbus-Distribution can be installed using+the standard Cabal way:++$ runhaskell Setup.hs configure+$ runhaskell Setup.hs build+$ runhaskell Setup.hs install --global # with root privileges++This will generate the library and the PortRegestry program.++For those who prefer to build it the old way with make:++$ make build+$ make install # with root privileges++If you want to run your own distributed system, you'll need to compile the+PortRegistry, too. This can be done with++$ make programs+++Steps to make a distributed system running+--------------------------------++Before you can use the mailboxes in your own programs, you need to start the+PortRegistry BEFORE running any of you programs. If you have compiled the+PortRegistry, you can start it with: ++$ cd Programs/PortRegistry+$ ./PortRegistry++This will create a file "registry.xml" in your "/tmp" directory. This file+contains all information for your programs to access the PortRegistry.+It is wise to copy this file to every computer, on which you want to run a +program of your system.++Before you can send/receive messages in your programs, you'll need to+load the "registry.xml" file. This can be done by two simple lines in the+IO-monad:++p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+setPortRegistry p++To create a simple Mailbox for receiving messages, you can look a the following+demo-program:+      +main :: IO ()+main+  = do+    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"+    setPortRegistry reg+    gS <- (newGlobalStream "global"):: IO (Stream String)+    msg <- readStream gS+    putStrLn msg++This program will set up a mailbox and wait till the first message is received+and print it out. Then is will terminate.+And the following program will send a message to your mailbox:      +      +main :: IO ()+main+  = do+    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"+    setPortRegistry reg+    gP <- (newGlobalPort "global")::IO (Port String)+    send gP "Hello World"+      +This program just sends one message to your mailbox.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ source/Holumbus/Common/Debug.hs view
@@ -0,0 +1,24 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Common.Debug+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  A typeclass for printing debug output.++-}+-- ----------------------------------------------------------------------------++module Holumbus.Common.Debug where+++class Debug m where++  -- | Just print out some debug output.  +  printDebug :: m -> IO ()
+ source/Holumbus/Common/FileHandling.hs view
@@ -0,0 +1,213 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Common.FileHandling+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  Some nice helper functions for strict file writing, appending and+  reading. All read-functions are strict, in the sense that+  the whole file is read at once and the content is stored into memory.+  So you won't have any semi-closed handles which might bring you in trouble+  from time to time. The files are immedialty closed after the reading.+  This Module can handle four filetypes: XML-Files, List-Files,+  Binary-Files and Text-Files.+  +  XML-Files:+  Normal textfiles, but the content is stored as a xml-representation. For+  the pickling, we use the HXT-Library +  (see http:\/\/www.fh-wedel.de\/~si\/HXmlToolbox\/)+  +  List-Files:+  Binary-Files, which store a list of data-objects. The main difference+  between Binary-Files and List-Files is, that you can append data to a+  List-File which will be automatically concatenated to the existing list.+  If you try this with a normal Binary-File, you'll get only the original+  list and the appended data will be lost.+  +  Binary-Files:+  Normal binary files, for the encoding and decoding, we use the Haskell+  binary-package.+  +  Text-Files:+  Normal text files.++-}+-- ----------------------------------------------------------------------------++module Holumbus.Common.FileHandling+    (+      -- * xml files+      loadFromXmlFile+    , saveToXmlFile++      -- * lists in binary files+    , writeToListFile+    , appendToListFile+    , readFromListFile+    , parseByteStringToList++      -- * bytestring file handling+    , writeToBinFile+    , appendToBinFile+    , readFromBinFile++      -- * string file handling+    , writeToTextFile+    , appendToTextFile+    , readFromTextFile+    )+where+++import           Control.Exception+import           Data.Binary+import qualified Data.ByteString.Lazy as B+import           Data.Char+import           Foreign+import           System.IO+import           System.IO.Unsafe++import           Text.XML.HXT.Arrow++-- ----------------------------------------------------------------------------+-- xml files+-- ----------------------------------------------------------------------------+++-- | Loads an XML-File and parses the content to a specified datatype.+loadFromXmlFile :: (XmlPickler a) => FilePath -> IO a+loadFromXmlFile f+  = do+    r <- runX (xunpickleDocument xpickle options f)+    return $! head r+    where+    options = [ (a_validate,v_0), (a_remove_whitespace, v_1), (a_encoding, utf8), (a_validate, v_0) ]+++-- | Converts a dataset to XML and saves the XML-string into a file.+saveToXmlFile :: (XmlPickler a) => FilePath -> a -> IO ()+saveToXmlFile f i +  = do+    runX (constA i >>> xpickleDocument xpickle options f)+    return ()+    where+    options = [ (a_indent, v_1), (a_output_encoding, utf8), (a_validate, v_0) ]++-- ----------------------------------------------------------------------------+-- generic lists+-- ----------------------------------------------------------------------------+++-- | Writes data to a list file.+writeToListFile :: (Binary a) => FilePath -> [a] -> IO ()+writeToListFile fp bs = writeToBinFile fp $ B.concat $ map encode bs+++-- | Appends data to a list file.+appendToListFile :: (Binary a) => FilePath -> [a] -> IO ()+appendToListFile fp bs = appendToBinFile fp $ B.concat $ map encode bs++-- | reads from a list file.+readFromListFile :: (Binary a) => FilePath -> IO [a]+readFromListFile f+   = do+     b <- readFromBinFile f +     return $ parseByteStringToList b+++-- | You'll need this function, if you read the files a a normal binary file,+--   but the content itself is a list. This function encodes the bytestring+--   into a list of the specified datatype.+parseByteStringToList :: (Binary a) => B.ByteString -> [a]+parseByteStringToList b = reverse $ parse b []+  where+  parse :: (Binary a) => B.ByteString -> [a] -> [a]+  parse bs accu+    | (B.null bs) = accu+    | otherwise   = parse (B.drop count bs) ([nextElem] ++ accu) +    where+    nextElem = decode bs+    count    = B.length (encode nextElem)+     +++-- ----------------------------------------------------------------------------     +-- strict functions, bytestrings only     +-- ----------------------------------------------------------------------------++-- | Write data to a binary file.+writeToBinFile :: FilePath -> B.ByteString -> IO ()+writeToBinFile = B.writeFile+++-- | Append data to a binary file, this doesn't mean, that the contents+--   are really concatenated when you read the file.    +appendToBinFile :: FilePath -> B.ByteString -> IO ()+appendToBinFile = B.appendFile+++-- | Reads the data from a binary file as a bytestring.+readFromBinFile :: FilePath -> IO B.ByteString+readFromBinFile f  +   = bracket (openBinaryFile f ReadMode) hClose $ +       \h -> do+       s <- hFileSize h+       c <- B.hGetNonBlocking h (fromInteger s)+       return $! c    +++-- ----------------------------------------------------------------------------     +-- strict functions, strings only     +-- ----------------------------------------------------------------------------++     +-- | Writes a sting to a text file.+writeToTextFile :: FilePath -> String -> IO ()+writeToTextFile = writeFile++-- | Appends a string to a text file.+appendToTextFile :: FilePath -> String -> IO ()+appendToTextFile = appendFile+++-- | Strict file reading by Simon Marlow.+-- found on+-- http:\/\/users.aber.ac.uk\/afc\/stricthaskell.html+readFromTextFile :: FilePath -> IO String+readFromTextFile f+  = do+    bracket (openFile f ReadMode) hClose $ +      \h -> do+      s <- hFileSize h+      fp <- mallocForeignPtrBytes (fromIntegral s)+      len <- withForeignPtr fp $ \buf -> hGetBuf h buf (fromIntegral s)+      lazySlurp fp 0 len+++buf_size :: Int+buf_size = 4096+++lazySlurp :: ForeignPtr Word8 -> Int -> Int -> IO String+lazySlurp fp ix len+  | fp `seq` False = undefined+  | ix >= len = return []+  | otherwise = do+      cs <- unsafeInterleaveIO (lazySlurp fp (ix + buf_size) len)+      ws <- withForeignPtr fp $ \p -> loop' (min (len-ix) buf_size - 1) +          ((p :: Ptr Word8) `plusPtr` ix) cs+      return ws+  where+  loop' :: Int -> Ptr Word8 -> String -> IO String+  loop' len' p acc+    | len' `seq` p `seq` False = undefined+    | len' < 0 = return acc+    | otherwise = do+       w <- peekElemOff p len'+       loop' (len'-1) p (chr (fromIntegral w):acc)
+ source/Holumbus/Common/Logging.hs view
@@ -0,0 +1,66 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Common.Logging+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  In this part of the Holumbus-Framework, we use hslogging. With this, we are+  able to activate and deactivate some nice debugging-Output.+  This is the global configuration for the logging-output. Of cource, if you+  want to use this output, you have to invoke initializeLogging.++-}+-- ----------------------------------------------------------------------------++module Holumbus.Common.Logging+(+-- * Configuration+initializeLogging+)+where++import System.IO++import System.Log.Logger+import System.Log.Handler.Simple+-- import System.Log.Handler.Syslog+++++-- ----------------------------------------------------------------------------+-- Configuration+-- ----------------------------------------------------------------------------+++-- | configures the logging-parameters for the Holumbus-Framework+initializeLogging :: [(String, Priority)] -> IO ()+initializeLogging ls = do++    -- log all verbose	+    v <- verboseStreamHandler stderr DEBUG+    updateGlobalLogger rootLoggerName (setLevel DEBUG . setHandlers [v])++    -- set all logLevels for all loggers+    mapM_ (\(s,p) -> updateGlobalLogger s (setLevel p)) ls++    -- log all to syslog+    -- s <- openlog "SyslogStuff" [PID] USER DEBUG+    -- updateGlobalLogger rootLoggerName (addHandler s)+    +    -- this will be removed in next versions+    -- updateGlobalLogger "Holumbus" (setLevel DEBUG)+    -- updateGlobalLogger "Holumbus.Network" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.MapReduce.TaskProcessor.task" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.FileSystem.Storage.FileStorage" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.FileSystem.Node.NodeData" (setLevel INFO)+    -- updateGlobalLogger "Holumbus.MapReduce.Types" (setLevel INFO)+    -- updateGlobalLogger "Holumbus.Network.Communication" (setLevel DEBUG)+    -- updateGlobalLogger "Holumbus.MapReduce.JobController.cycle" (setLevel WARNING)+
+ source/Holumbus/Common/Threading.hs view
@@ -0,0 +1,206 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Common.Threading+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  Operations to start and stop threads which will not be killed when a regular+  exception occurs. In this case, the thread will continue working. Such a +  thread can only be killed by the stop-method. This whole thing is a wrapper+  around the normal lightweight thread functions.+  +  The created threads execute a function in an infinite loop. This is the+  normal usecase for message dispatcher threads.+  +-}++-- ----------------------------------------------------------------------------++module Holumbus.Common.Threading+(+  Thread+, newThread+, setThreadDelay+, setThreadAction+, setThreadErrorHandler++, startThread+, stopThread+)+where++{- 6.8+import qualified Control.Exception      as E+import           Data.Typeable+-}+import           Control.Exception      ( AsyncException(..)+                                        , catchJust+                                        )+import           Control.Concurrent++import           Data.Maybe++import           System.Log.Logger++import           Holumbus.Common.Utils  ( handleAll )++localLogger :: String+localLogger = "Holumbus.Common.Threading"+++-- ----------------------------------------------------------------------------+-- Thread-Control+-- ----------------------------------------------------------------------------++-- | Exception to stop a thread, so we can distinguish this request from+--   "real" exceptions.++{- 6.8+data KillThreadException = KillThreadException ThreadId+  deriving (Typeable, Show)+-}++-- | The data needed to access and control the thread. +data ThreadData = ThreadData {+    thd_Id      :: Maybe ThreadId+  , thd_Running :: Bool+  , thd_Delay   :: Maybe Int+  , thd_Action  :: (IO ())+  , thd_Error   :: (IO ())+  }+  +  +-- | The thread datatype+type Thread = MVar ThreadData +++-- | Creates a new thread object. The thread will not be running.+newThread :: IO Thread+newThread+  = do +    newMVar $ ThreadData Nothing False defaultDelay noAction noAction+    where+    noAction     = return ()+    defaultDelay = Nothing+++-- | Sets the delay between two loop cycles. Default value: no delay.+setThreadDelay :: Int -> Thread -> IO ()+setThreadDelay d thread+  = do+    modifyMVar thread $ \thd -> return (thd {thd_Delay = Just d},())+    +    +-- | Sets the action function, which will be executed in each cycle+setThreadAction :: (IO ()) -> Thread -> IO ()+setThreadAction f thread+  = do+    modifyMVar thread $ \thd -> return (thd {thd_Action = f},())+++-- | Sets the error handler. It is activated, when the action function+--   will raise an exception.+setThreadErrorHandler :: (IO ()) -> Thread -> IO ()+setThreadErrorHandler e thread+  = do+    modifyMVar thread $ \thd -> return (thd {thd_Error = e},())+++-- | Starts the thread.+startThread :: Thread -> IO ()+startThread th+  = modifyMVar th $+      \thd ->+      do+      -- check, if thread is running (it has a threadId)+      servId' <- case (thd_Id thd) of+        i@(Just _) -> return i+        (Nothing) ->+          do+          -- if not running, start the loop+          i <- forkIO $ doAction th+          return (Just i)+      return (thd {thd_Id = servId', thd_Running = True},())+    where    +    -- the action loop+    doAction thread+      = do+        thd <- readMVar thread+        if (thd_Running thd)+          -- if thread is still running +          then do+            {- 6.8 E.handle -}+            handleAll+             ( \ e -> do+                      -- if a normal exception occurs, the error handler should be excuted+                      warningM localLogger $ show e+                      thd_Error thd+                      doAction thread+             ) $+              do+              -- catch the exception which tells us to kill the dispatcher and kill it+              {- 6.8 E.catchDyn -}+              catchJust isThreadKilledException+               ( do+                 -- wait for next watch-cycle+                 if (isJust (thd_Delay thd)) +                    then do threadDelay $ fromJust (thd_Delay thd)+                    else do return ()+                 -- do the action+                 thd_Action thd+                 -- and again+                 doAction thread+               )+               killHandler+          else do+            debugM localLogger $ "thread normally closed by himself"+            deleteThreadId thread+	where+        isThreadKilledException      		:: AsyncException -> Maybe ()+        isThreadKilledException ThreadKilled    = Just ()+        isThreadKilledException _               = Nothing++        killHandler :: () -> IO ()+        killHandler _+          = do+            debugM localLogger $ "thread normally closed by other thread "+            deleteThreadId thread+{- 6.8+        killHandler :: KillThreadException -> IO ()+        killHandler (KillThreadException i)+          = do+            debugM localLogger $ "thread normally closed by other thread " ++ show i+            deleteThreadId thread+-}+        deleteThreadId :: Thread -> IO ()+        deleteThreadId t+          = modifyMVar t $ \thd -> return (thd {thd_Id = Nothing, thd_Running = False},())+++-- | Stops the thread. If the thread itself wants to stop from within the action+--   function, the current cycle will be executed till the end. So statements+--   after this function will still be executed.+stopThread :: Thread -> IO ()+stopThread thread+  = do+    me <- myThreadId+    him <- withMVar thread $ \thd -> return $ thd_Id thd+    if (isJust him) +      then do+        let he = fromJust him+        if (me == he) +          -- if stop is called from the thread, we want to finish our work+          then do +            modifyMVar thread $ \thd -> return (thd {thd_Running = False},())+            -- else we kill it the unfriendly way...+          else do+            {- 6.8: E.throwDynTo he (KillThreadException me) -}+            killThread he+      else do+        return ()
+ source/Holumbus/Common/Utils.hs view
@@ -0,0 +1,134 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Common.Utils+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  Some nice functions, needed everywhere.++-}++-- ----------------------------------------------------------------------------++module Holumbus.Common.Utils+    ( +      decodeMaybe++      -- , lookupList+      -- , lookupMaybe++      -- , cutMaybePair+    , filterEmptyList+    , setEmptyList+      -- , filterBlanks++    , prettyRecordLine+      -- , prettyResultLine+      -- , handleExitError++    , handleAll		-- exception handling for all exceptions+    )+where++import           Control.Exception	( SomeException+					, handle+					)+import           Data.Binary+import qualified Data.ByteString.Lazy as B++-- import qualified Data.Map as Map++import           Data.Maybe+import           Data.Char++-- import           System.Exit++-- ------------------------------------------------------------+--+-- handle with type sinature++{- 6.8+handleAll	:: (Exception -> IO a) -> IO a -> IO a+-}++-- ghc 6.10 requires a type signature for handle++handleAll	:: (SomeException -> IO a) -> IO a -> IO a+handleAll	= handle++-- ------------------------------------------------------------++-- | parses something from a maybe bytestring, if Nothing, then Nothing+decodeMaybe :: (Binary a) => Maybe B.ByteString -> Maybe a+decodeMaybe Nothing = Nothing+decodeMaybe (Just b) = (Just $ decode b)++{-+lookupMaybe :: (Ord k) => Map.Map k v -> Maybe k -> Maybe v+lookupMaybe _ Nothing = Nothing+lookupMaybe m (Just k) = Map.lookup k m+-}+{-+lookupList :: (Ord k) => Map.Map k v -> [k] -> [v]+lookupList m ks = mapMaybe (\k' -> lookupMaybe m $ Just k') ks+-}    +{-+cutMaybePair :: Maybe (Maybe a, Maybe b) -> (Maybe a, Maybe b)+cutMaybePair (Nothing) = (Nothing, Nothing)+cutMaybePair (Just p)  = p+-}++filterEmptyList :: Maybe [k] -> [k]+filterEmptyList (Nothing) = []+filterEmptyList (Just ls) = ls+++setEmptyList :: [k] -> Maybe [k]+setEmptyList [] = Nothing+setEmptyList ls = Just ls ++{-+filterBlanks :: Maybe String -> Maybe String+filterBlanks (Nothing) = Nothing+filterBlanks (Just "") = Nothing+filterBlanks (Just s)  = if s' == "" then Nothing else (Just s)+  where+  s' = filter isPrint s +-}+++-- | For the nice output of key-value-pairs +prettyRecordLine :: (Show b, Show a) => Int -> a -> b -> String+prettyRecordLine n a b = stra ++ (replicate diff ' ') ++ show strb+  where+   stra = show a+   strb = show b+   diff = n - length stra+  +   +{-+prettyResultLine :: (Show e, Show a, Show d) => Int -> Either e d -> a -> String+prettyResultLine n (Left e) _ = prettyRecordLine n "Error:" e+prettyResultLine n (Right r) a = prettyRecordLine n a r+-}++{-+handleExitError :: (Show e) => IO (Either e a) -> IO a+handleExitError res+  = do+    res' <- res+    case res' of+      (Left e) ->+        do+        putStrLn $ "ERROR: " ++ show e+        exitFailure+      (Right a) -> return a+-}   ++-- ----------------------------------------------------------------------------
+ source/Holumbus/Console/Console.hs view
@@ -0,0 +1,242 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Console.Console+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module provides a tiny and nice implementation of a little command +  shell. It can be feed with individual commands and provides a simple but+  powerful way to interact with your program. The following functions are+  implemented by default:+    exit - exit the console loop+    help - print a nice help++  There was a little "bug" with the System.Console.Readline package. When+  we use this option, we make a foreign call... and the Haskell library+  documentation say this about concurrency and GHC:+  +  "If you don't use the -threaded option, then the runtime does not make +  use of multiple OS threads. Foreign calls will block all other running +  Haskell threads until the call returns.+  The System.IO library still does multiplexing, so there can be multiple+  threads doing IO, and this is handled internally by the runtime using +  select."+   +  We make a foreign call, which is not in the System.IO library, so we+  have to work with -threaded when we want a fancy command history.++-}++-- ----------------------------------------------------------------------------+++module Holumbus.Console.Console+    (+     -- * Console datatype+     ConsoleData+    +     -- * Operations+    , nextOption+    , parseOption+    , initializeConsole+    , addConsoleCommand+    , handleUserInput+    )+where++import           Control.Concurrent+import           Control.Monad++import           Data.Char+import qualified Data.Map as Map++import           System.IO+import           System.Console.Editline.Readline++import           Holumbus.Common.Utils	( handleAll )++-- ----------------------------------------------------------------------------+-- datatypes+-- ----------------------------------------------------------------------------+++-- | Map which contains all commands that the user can execute+type ConsoleData a = Map.Map String (ConsoleCommand a)+++-- | Console command, only a pair of a function which will be executed +--   and a description +type ConsoleCommand a = ( Maybe (ConsoleFunction a), String )+++-- | A console function. The string list represents the arguments+type ConsoleFunction a = (a -> [String] -> IO())++-- ----------------------------------------------------------------------------+-- operations +-- ----------------------------------------------------------------------------+++-- | Creates a new console datatype+initializeConsole :: ConsoleData a+initializeConsole = Map.fromList [(exitString, exitCommand), (helpString, helpCommand)]+++-- | Adds a new console command to the function, an existing command with the+--   same name will be overwritten+addConsoleCommand +  :: String             -- ^ command string (the word the user has to enter when he wants to execute the command)+  -> ConsoleFunction a  -- ^ the function which should be executed+  -> String             -- ^ the function description+  -> ConsoleData a      -- ^ the old console data+  -> ConsoleData a+addConsoleCommand c f d m = Map.insert c (Just f, d) m+++-- | The exit function string.+exitString  :: String+exitString = "exit"+++-- | A dummy exit function (Just to print the help description, the command +--   is handled in the main loop.+exitCommand :: ConsoleCommand a+exitCommand = ( Nothing, "exit the console")+++-- | The help function string.+helpString :: String+helpString = "help"+++-- | A dummy help function (Just to print the help description, the command +--   is handled in the main loop.+helpCommand :: ConsoleCommand a+helpCommand = (Nothing, "print this help")+++-- | The command-line prompt string.+shellString :: String+shellString = "command> "++-- | gets the next option from the command line as string+nextOption :: [String] -> IO (Maybe String, [String])+nextOption o+  = handleAll (\_ -> return (Nothing, o)) $+      do+      if ( null o ) then+          return (Nothing, o)+        else +          return (Just $ head o, tail o) ++-- | Simple "parser" for the commandline...+parseOption :: Read a => [String] -> IO (Maybe a, [String])+parseOption o+  = handleAll (\_ -> return (Nothing, o)) $+      do+      if ( null o ) then+          return (Nothing, o)+        else +          return (Just $ read $ head o, tail o) +++-- | Reads the input from the stdin.+getCommandLine :: IO (String)+getCommandLine +-- no commandline history+{-  = do+    putStr shellString+    hFlush stdout+    maybeLine <- getLine --readline shellString+    return maybeLine+-}+-- with commandline history, please use -threaded option+  = do+    maybeLine <- readline shellString+    yield+    case maybeLine of+      Nothing   -> return exitString -- EOF / control-d+      Just line -> +        do +          if (not $ null line) then do addHistory line else return ()+          return line+          ++-- | The main loop. You know... read stdin, parse the input, execute command.+--   You can quit it by the exit-command.+handleUserInput :: ConsoleData a -> a -> IO ()+handleUserInput cdata conf+  = do+    line  <- getCommandLine+    input <- return (words line)+    cmd   <- return (command input)+    args  <- return (arguments input)+    if (cmd == exitString) +      then do+        return ()+      else do+        if (not $ null cmd) then do handleCommand cdata conf cmd args else return ()+        handleUserInput cdata conf+    where+      command s = if (not $ null s) then head s else ""+      arguments s = tail s+++-- | Picks the command an execute the command function.+handleCommand :: ConsoleData a -> a -> String -> [String] -> IO ()+handleCommand cdata conf cmd args+  = do+    if (cmd == helpString)+      then do+        printHelp cdata+      else do+        handleCommand' (Map.lookup cmd cdata)+    where+        handleCommand' Nothing              = do printError+        handleCommand' (Just (Nothing, _ )) = do printNoHandler+        handleCommand' (Just (Just f, _ ))  = do f conf args+++-- | Is executed when the function has no handler+printNoHandler :: IO ()+printNoHandler +  = do+    putStrLn "no function handler found"+++-- | Prints the "command-not-found" message.+printError :: IO ()+printError+  = do+    putStrLn "unknown command, try help for a list of available commands"+++-- | Prints the help text.+printHelp :: ConsoleData a -> IO ()+printHelp cdata+  = do+    putStrLn "available Commands:"+    printCommands (Map.toAscList cdata)+    where+      printCommands [] = do return ()+      printCommands (x:xs) = do+                             printCommand x+                             printCommands xs+      printCommand (c, (_, t)) = do+                                 putStrLn ((prettyCommandName 15 c) ++ " -  " ++ t) +++-- | Does some pretty printing for the function descriptions+prettyCommandName :: Int -> String -> String +prettyCommandName n s+  | n <= 0 = s+  | (n > 0) && (null s) = ' ' : prettyCommandName (n-1) s+  | otherwise           = x : prettyCommandName (n-1) xs+    where+      (x:xs) = s
+ source/Holumbus/Data/AccuMap.hs view
@@ -0,0 +1,122 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Data.AccuMap+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  A map of key-value-pairs. The values are hold in a list, so adding the same+  key-value-pair twice to the map, will result in storing the value twice.+  Even the sequence of adding the values will be saved.+  +  The name AccuMap is from accumulation map. You can use this map to easily+  storing all you key-value-pairs. After that you can all values-lists by+  key. Unlike the MultiMap, you don't lose the information of identical values+  and their order of adding.+  +  Most functions are inspired by the Data.Map type. +-}+-- ----------------------------------------------------------------------------++module Holumbus.Data.AccuMap+(+  AccuMap+, empty+, null+, insert+, insertList+, lookup+, member+, deleteKey+, union+, fromList+, fromTupleList+, toList+)+where++import           Prelude hiding (null, lookup)++import qualified Data.Map as Map+++-- | the AccuMap datatype+data AccuMap k a = AM (Map.Map k [a])+  deriving (Eq, Ord)+++instance (Show k, Show a) => Show (AccuMap k a) where+  show (AM m) = msShow+    where+    ms = Map.toList m+    msShow = concat $ map (\(k,s) -> (show k) ++ "\n" ++ (showL s)) ms+    showL ls = concat $ map (\s -> show s ++ "\n") ls +++-- | Creates an empty AccuMap.+empty :: (Ord k) => AccuMap k a+empty = AM Map.empty+++-- | Test, if AccuMap is empty.+null :: (Ord k) => AccuMap k a -> Bool+null (AM m) = Map.null m+++-- | Insert a key-value-pair to the AccuMap.+insert :: (Ord k) => k -> a -> AccuMap k a -> AccuMap k a+insert k a (AM m) = AM $ Map.alter altering k m+  where+    altering Nothing = Just $ [a]+    altering (Just s) = Just $ a:s+++-- | Insert a key and a list of values to the AccuMap.+--   Faster than adding all pair one by one.+insertList :: (Ord k) => k -> [a] -> AccuMap k a -> AccuMap k a+insertList _ [] m      = m+insertList k as (AM m) = AM $ Map.alter altering k m+  where+    altering Nothing = Just $ as+    altering (Just s) = Just $ as ++ s+++-- | Get the list of values for one key. If the key doesn't exists,+--   an empty list is returned.+lookup :: (Ord k) => k -> AccuMap k a -> [a]+lookup k (AM m) = maybe [] (id) (Map.lookup k m)+++-- | Test, if the key is in the AccuMap.+member :: (Ord k, Eq a) => k -> AccuMap k a -> Bool+member k m = [] /= lookup k m+++-- | Deletes a key and all its values from the AccuMap.+deleteKey :: (Ord k) => k -> AccuMap k a -> AccuMap k a+deleteKey k (AM m) = AM $ Map.delete k m+++-- | Combines two AccuMaps, the ordering of the two maps is+--   significant for the order of the values-elements.+union :: (Ord k) => AccuMap k a -> AccuMap k a -> AccuMap k a+union (AM m1) (AM m2) = AM $ Map.unionWith (\l1 l2 -> l1 ++ l2) m1 m2 +++-- | Creates an AccuMap from a list.+fromList :: (Ord k) => [(k,[a])] -> AccuMap k a+fromList ks = foldl (\m (k,as) -> insertList k as m) empty ks+++-- | Creates an AccuMap from a tuple list.+fromTupleList :: (Ord k) => [(k,a)] -> AccuMap k a+fromTupleList ts = fromList $ map (\(k,a) -> (k,[a])) ts+++-- | Transforms an AccuMap to a list.+toList :: (Ord k) => AccuMap k a -> [(k,[a])]+toList (AM m) = Map.toList m
+ source/Holumbus/Data/KeyMap.hs view
@@ -0,0 +1,127 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Data.KeyMap+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  The KeyMap is derived from the Data.Map type. The keys of the Map are+  strings and the values can be arbitrary data objects. But you don't have to+  specify the keys because every value-object is able to create it's own+  key via the "Key" typeclass.+  +  From the functionality, the KeyMap stands between a set and a Map. If you+  want to insert an element to the map, it behaves like a set. You don't need +  an additionnal key and it makes no different if you insert an object multiple+  times. If you want to access the objects in the KeyMap, you can lookup them+  via the key, so in this case this container behaves like an ordinary map.++  The functions for this container are named after the standard Map and Set+  functions.+-}+-- ----------------------------------------------------------------------------++module Holumbus.Data.KeyMap+    (+      KeyMap+    , Key(..)+    , empty+    , null+    , insert+    , lookup+    , keys+    , elems+    , memberKey+    , memberElem+    , deleteKey+    , deleteElem+    , fromList+    , toList+    , toAscList+    )+where++import           Prelude hiding (null, lookup)+import           Data.Maybe+import qualified Data.Map as Map++-- | Every element of this map has to implement a key-function. which+--   gives us the key of the element+class Key n where+  getKey :: n -> String+++-- | The KeyMap datatype.+data KeyMap a = NM (Map.Map String a)+  deriving (Show, Eq, Ord)+++-- | The empty KeyMap.+empty :: KeyMap a+empty = NM Map.empty+++-- | Test, if the MultiMap is empty.+null :: (Key a) => KeyMap a -> Bool+null (NM m) = Map.null m+++-- | Inserts an element in the KeyMap.+insert :: (Key a) => a -> KeyMap a -> KeyMap a+insert a (NM m) = NM $ Map.insert (getKey a) a m+  ++-- | Gets all different elements for one key or an empty set.+lookup :: (Key a, Monad m) => String -> KeyMap a -> m a+lookup k (NM mm)+    = do+      let Just r = Map.lookup k mm+      return r++-- | Get all different keys from the map.+keys :: (Key a) => KeyMap a -> [String]+keys (NM m) = Map.keys m+++-- | Get all different values in the map without regarding their keys.+elems :: (Key a) => KeyMap a -> [a]+elems (NM m) = Map.elems m+++-- | Test, if a key is in the KeyMap.+memberKey :: (Key a) => String -> KeyMap a -> Bool+memberKey k (NM m) = Map.member k m+++-- | Test, if a data object is in the KeyMap.+memberElem :: (Key a) => a -> KeyMap a -> Bool+memberElem a (NM m) = Map.member (getKey a) m+++-- | Deletes a whole key from the KeyMap.+deleteKey :: (Key a) => String -> KeyMap a -> KeyMap a+deleteKey k (NM m) = NM $ Map.delete k m+++-- | Deletes a single elemet from the KeyMap.+deleteElem :: (Key a) => a -> KeyMap a -> KeyMap a+deleteElem a (NM m) = NM $ Map.delete (getKey a) m+++-- | Creates a KeyMap from a list of keys.+fromList :: (Key a) => [a] -> KeyMap a+fromList as = foldl (\m a -> insert a m) empty as+++-- | Transforms a KeyMap to a list of keys.+toList :: (Key a) => KeyMap a -> [a]+toList (NM m) = map (snd) $ Map.toList m+++-- | The same as toList, but the keys are in ascending order.+toAscList :: (Key a) => KeyMap a -> [a]+toAscList (NM m) = map (snd) $ Map.toAscList m
+ source/Holumbus/Data/MultiMap.hs view
@@ -0,0 +1,178 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Data.MultiMap+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module provides a MultiMap, that means a Map, which can hold+  multiple values for one key, but every distinct value is only stores once.+  So adding the same key-value-pair twice will only create one new entry in+  the map.+  +  This Map is helpfull to examine how many different key-values-pairs you+  have in your application.+  +  Most of the functions are borrowed from Data.Map +-}+-- ----------------------------------------------------------------------------++module Holumbus.Data.MultiMap+(+  MultiMap+, empty+, null+, insert+, insertSet+, insertKeys+, lookup+, keys+, elems+, filterElements+, member+, delete+, deleteKey+, deleteElem+, deleteAllElems+, fromList+, fromTupleList+, toList+, toAscList+)+where++import           Prelude hiding (null, lookup)++import qualified Data.Map as Map+import qualified Data.Set as Set+++-- | A MultiMap, it can hold more (different!!!) Elements for one key.+data MultiMap k a = MM (Map.Map k (Set.Set a))+  deriving (Show, Eq, Ord)++{-+instance (Show k, Show a) => Show (MultiMap k a) where+  show (MM m) = msShow+    where+    ms = map (\(k,s) -> (k, Set.toList s)) (Map.toList m)+    msShow = concat $ map (\(k,s) -> (show k) ++ "\n" ++ (showL s)) ms+    showL ls = concat $ map (\s -> show s ++ "\n") ls +-}++-- | The empty MultiMap.+empty :: (Ord k, Ord a) => MultiMap k a+empty = MM Map.empty+++-- | Test, if the MultiMap is empty.+null :: (Ord k, Ord a) => MultiMap k a -> Bool+null (MM m) = Map.null m+++-- | Inserts an element in the MultiMap.+insert :: (Ord k, Ord a) => k -> a -> MultiMap k a -> MultiMap k a+insert k a (MM m) = MM $ Map.alter altering k m+  where+  altering Nothing = Just $ Set.singleton a+  altering (Just s) = Just $ Set.insert a s+  ++-- | Inserts multiple elements in a set to the MultiMap.+insertSet :: (Ord k, Ord a) => k -> Set.Set a -> MultiMap k a -> MultiMap k a+insertSet k newSet mm@(MM m) = +  if (Set.null newSet) then mm else MM $ Map.alter altering k m+  where+  altering Nothing = Just newSet+  altering (Just s) = Just $ Set.union newSet s+++-- | Inserts multiple keys with the same values.+insertKeys :: (Ord k, Ord a) => [k] -> Set.Set a -> MultiMap k a -> MultiMap k a+insertKeys ks a m = foldl (\m' k -> insertSet k a m') m ks +++-- | Gets all different elements for one key or an empty set.+lookup :: (Ord k, Ord a) => k -> MultiMap k a -> Set.Set a+lookup k (MM m) = maybe (Set.empty) (id) (Map.lookup k m)+++-- | Get all different elements from a list of keys.+lookupKeys :: (Ord k, Ord a) => [k] -> MultiMap k a -> Set.Set a+lookupKeys ks m = Set.unions $ map (\k -> lookup k m) ks+++-- | Get all different keys from the map.+keys :: (Ord k, Ord a) => MultiMap k a -> Set.Set k+keys (MM m) = Set.fromList $ Map.keys m+++-- | Get all different values in the map without regarding their keys.+elems :: (Ord k, Ord a) => MultiMap k a -> Set.Set a+elems (MM m) = Set.unions $ Map.elems m+++-- | Like lookup keys, but an empty input list will give all elements back,+--   not the empty set.+filterElements :: (Ord k, Ord a) => [k] -> MultiMap k a -> Set.Set a+filterElements [] m = elems m  -- get all+filterElements ks m = lookupKeys ks m+++-- | Test, if a key is in the Map.+member :: (Ord k, Ord a) => k -> MultiMap k a -> Bool+member k m = Set.empty /= lookup k m+++-- | Deletes an Element from the Map, if the data in Nothing, the whole key is+--   deleted.+delete :: (Ord k, Ord a) => k -> Maybe a -> MultiMap k a -> MultiMap k a+delete k Nothing m = deleteKey k m+delete k (Just a) m = deleteElem k a m+++-- | Deletes a whole key from the map.+deleteKey :: (Ord k, Ord a) => k -> MultiMap k a -> MultiMap k a+deleteKey k (MM m) = MM $ Map.delete k m+++-- | Deletes a single Element from the map.+deleteElem :: (Ord k, Ord a) => k -> a -> MultiMap k a -> MultiMap k a+deleteElem k a (MM m) = MM $ Map.alter delSet k m+  where+  delSet Nothing = Nothing+  delSet (Just set) = filterEmpty $ Set.delete a set+  filterEmpty set+    | set == Set.empty = Nothing+    | otherwise = Just set+++-- | Deletes all Elements (*,a) (slow!!!).+deleteAllElems :: (Ord k, Ord a) => a -> MultiMap k a -> MultiMap k a+deleteAllElems a m = foldl (\m'' k -> deleteElem k a m'') m ks+  where+  ks = Set.toList $ keys m+++-- | Creates a MultiMap from a list of pairs (key,set value).+fromList :: (Ord k, Ord a) => [(k,Set.Set a)] -> MultiMap k a+fromList ks = foldl (\m (k,as) -> insertSet k as m) empty ks+++-- | Creates a MultiMap from a list of tuples.+fromTupleList :: (Ord k, Ord a) => [(k,a)] -> MultiMap k a+fromTupleList ks = foldl (\m (k,a) -> insert k a m) empty ks+++-- | Transforms a MultiMap to a list of pairs (key,set value).+toList :: (Ord k, Ord a) => MultiMap k a -> [(k,Set.Set a)]+toList (MM m) = Map.toList m+++-- | The same as toList, but the keys are in ascending order.+toAscList :: (Ord k, Ord a) => MultiMap k a -> [(k,Set.Set a)]+toAscList (MM m) = Map.toAscList m
+ source/Holumbus/Network/Communication.hs view
@@ -0,0 +1,959 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Network.Communication+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  This module implements an abstract client server model. The clients register+  at the server and check from time to time if the server still exists. If not,+  the client searches for a new server.+  The server on the other hand, keeps a list with all clients and checks, if+  each client is reachable. If not, the client is deleted from the list.+  +  This abstract network model helps us to implement a basic distrubuted system+  with a central server and many attached clients which get little tasks from+  the server. Because this model alone would be very unfunctional, the user is+  able to define his own functions which will be handled by the server or the+  client. +-}++-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Network.Communication+(+  StreamName       -- (reexport)+, SocketId         -- (reexport)+, PortNumber       -- (reexport)++-- time constants+, time30           -- (reexport)+, timeIndefinitely -- (reexport)++, IdType++, ClientInfo(..)++-- * server operations+, Server+, newServer+, closeServer+, ServerPort+, newServerPort+, sendRequestToServer++, getClientInfo+, getAllClientInfos++-- * client operations+, ClientClass(..)+, Client+, newClient+, closeClient+, ClientPort+, sendRequestToClient +)+where++import           Control.Concurrent+import           Data.Binary+import qualified Data.ByteString.Lazy as B+import qualified Data.Map as Map+import           Data.Maybe+import           Network+import           System.Log.Logger++import           Holumbus.Common.Debug+import           Holumbus.Common.Threading+import qualified Holumbus.Data.MultiMap as MMap+import           Holumbus.Network.Messages+import           Holumbus.Network.Port+import           Holumbus.Network.Site++++localLogger :: String+localLogger = "Holumbus.Network.Communication"+++-- ----------------------------------------------------------------------------+-- General Datatypes+-- ----------------------------------------------------------------------------++-- | The type of the client id.+type IdType = Int++++-- ----------------------------------------------------------------------------+-- Server-Messages+-- ----------------------------------------------------------------------------++-- | The requests, the server can handle.+data ServerRequestMessage+  = SReqRegisterClient SiteId (Port ClientRequestMessage)+  | SReqUnregisterClient IdType+  | SReqPing IdType+  | SReqServerAction B.ByteString+  | SReqUnknown+  deriving (Show)++instance Binary (ServerRequestMessage) where+  put (SReqRegisterClient sid po) = putWord8 1 >> put sid >> put po+  put (SReqUnregisterClient i)    = putWord8 2 >> put i+  put (SReqPing i)                = putWord8 3 >> put i+  put (SReqServerAction b)        = putWord8 4 >> put b+  put (SReqUnknown)               = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \sid -> get >>= \po -> return (SReqRegisterClient sid po)+        2 -> get >>= \i -> return (SReqUnregisterClient i)+        3 -> get >>= \i -> return (SReqPing i)+        4 -> get >>= \b -> return (SReqServerAction b)+        _ -> return (SReqUnknown)+++-- | The responses the server gives.+data ServerResponseMessage+  = SRspSuccess+  | SRspRegisterClient IdType+  | SRspUnregisterClient+  | SRspPing Bool+  | SRspServerAction B.ByteString+  | SRspError String+  | SRspUnknown+  deriving (Show)++instance Binary (ServerResponseMessage) where+  put (SRspSuccess)          = putWord8 1+  put (SRspRegisterClient i) = putWord8 2 >> put i+  put (SRspUnregisterClient) = putWord8 3+  put (SRspPing b)           = putWord8 4 >> put b+  put (SRspServerAction b)   = putWord8 5 >> put b+  put (SRspError e)          = putWord8 6 >> put e+  put (SRspUnknown)          = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> return (SRspSuccess)+        2 -> get >>= \i -> return (SRspRegisterClient i)+        3 -> return (SRspUnregisterClient)+        4 -> get >>= \b -> return (SRspPing b)+        5 -> get >>= \b -> return (SRspServerAction b)+        6 -> get >>= \e -> return (SRspError e)+        _ -> return (SRspUnknown)+  +instance RspMsg (ServerResponseMessage) where+  isError (SRspError _) = True+  isError _ = False+  +  getErrorMsg (SRspError e) = e+  getErrorMsg _ = ""+  +  isUnknown (SRspUnknown) = True+  isUnknown _ = False+  +  mkErrorMsg e = SRspError e+  ++++-- ----------------------------------------------------------------------------+-- Server-TypeClass+-- ----------------------------------------------------------------------------++-- | The request-functions a server has to implement.+class ServerClass s where++  -- | Register a new client in the server database.+  registerClient :: SiteId -> Port ClientRequestMessage -> s -> IO IdType+  +  -- | Delete a client from the server database.+  unregisterClient :: IdType -> s -> IO ()+  +  -- | Check, if server is responding.+  pingServer :: IdType-> s -> IO Bool++++-- ----------------------------------------------------------------------------+-- Server-Data+-- ----------------------------------------------------------------------------+  +-- | The type of the functions which will be executed by registration and+--   unregistration.+type RegistrationAction = (IdType -> ClientPort -> IO ())+  +  +-- The information of the client known by the server.+data ClientInfo = ClientInfo {+    ci_Id           :: Int+  , ci_Site         :: SiteId                -- ^ SiteId (Hostname,PID) of the client process+  , ci_Port         :: ClientPort            -- ^ the port of the client+  , ci_PingThreadId :: Thread                -- ^ the threadId of the ping-Process (needed to stop it)+  , ci_LifeValue    :: Int+  }+  +instance Show ClientInfo where+  show (ClientInfo n s p _ i) = "{Id: " ++ show n ++ +                                 " - Site: " ++ show s ++ +                                 " - Port: " ++ show p ++ +                                 " - LifeValue: " ++ show i ++ "}"+++-- | The data of the server needed to organise the clients.+data ServerData = ServerData {+    sd_ServerThreadId  :: Thread                        -- ^ threadId of the streamDispatcher+  , sd_OwnStream       :: Stream (ServerRequestMessage) -- ^ the stream the requestDispatcher reads from+  , sd_OwnPort         :: Port (ServerRequestMessage)   -- ^ the port the clients send messages to+  , sd_ClientMap       :: Map.Map IdType ClientInfo     -- ^ infomation of the the clients+  , sd_SiteToClientMap :: MMap.MultiMap SiteId IdType   -- ^ needed to get the closest client+  , sd_SiteMap         :: SiteMap                       -- ^ needed to get the closest site+  , sd_Register        :: RegistrationAction+  , sd_Unregister      :: RegistrationAction+  , sd_NextId          :: IdType+  }+  +  +-- | The server.+data Server = Server (MVar ServerData)+  +  +maxLifeValue :: Int+maxLifeValue = 3+++-- | Creates a new server.+newServer+  :: (Binary a, Binary b)+  => StreamName -> Maybe PortNumber+  -> (a -> IO (Maybe b))             -- ^ handling own request+  -> Maybe RegistrationAction        -- ^ for registration+  -> Maybe RegistrationAction        -- ^ for unregistration +  -> IO Server+newServer sn pn dispatch register unregister+  = do+    -- create a new server+    st    <- (newStream STGlobal (Just sn) pn::IO (Stream ServerRequestMessage))+    po    <- newPortFromStream st+    tid   <- newThread+    let reg   = maybe (\_ _ -> return ()) id register+    let unreg = maybe (\_ _ -> return ()) id unregister+    let sd = ServerData tid st po Map.empty MMap.empty Map.empty reg unreg 1+    s <- newMVar sd+    let server =  Server s+    -- start the requestDispatcher to handle requests+    startRequestDispatcher tid st (dispatchServerRequest server dispatch)+    return server+++-- | Closes the server.+closeServer :: Server -> IO ()+closeServer s@(Server server)+  = do+    debugM localLogger "closeServer: start"+    (allIds,thread,stream) <- withMVar server $+      \sd ->+      do+      -- getAll ClientIds+      let allIds = Map.keys (sd_ClientMap sd)+      return (allIds, sd_ServerThreadId sd, sd_OwnStream sd)          +    debugM localLogger "closeServer: stopRequestDispatcher"+    -- shutdown the server thread+    stopRequestDispatcher thread+    -- close the stream+    debugM localLogger "closeServer: closeStream"+    closeStream stream+    debugM localLogger "closeServer: unregister clients"+    mapM (\i -> do unregisterClient i s) allIds+    debugM localLogger "closeServer: end"+    return ()+++-- | Handles the requests from the client.+dispatchServerRequest+  :: (Binary a, Binary b)+  => Server+  -> (a -> IO (Maybe b))+  -> ServerRequestMessage+  -> Port (ServerResponseMessage)+  -> IO ()+dispatchServerRequest server action msg replyPort+  = do+    debugM localLogger $ "dispatchServerRequest: " ++ show msg+    case msg of+      (SReqRegisterClient s po) ->+        do+        handleRequest replyPort (registerClient s po server) (\i -> SRspRegisterClient i)+        return ()+      (SReqUnregisterClient n) ->+        do+        handleRequest replyPort (unregisterClient n server) (\_ -> SRspUnregisterClient)+        return ()+      (SReqPing i) ->+        do+        handleRequest replyPort (pingServer i server) (\b -> SRspPing b)+        return ()+      (SReqServerAction b) ->+        do+        handleRequest replyPort+          (action $ decode b) +          (\res -> maybe (SRspUnknown) (\r -> SRspServerAction $ encode r) res)+      _ -> +        handleRequest replyPort (return ()) (\_ -> SRspUnknown)+++-- | Creates a new client id and updates the serverdata.+getNextId :: ServerData -> (IdType, ServerData)+getNextId sd +  = (i, sd { sd_NextId = nid })+  where+    i   = sd_NextId sd+    nid = i + 1+++-- | Adds a new client to the server datastructures,+--   the ping-thread will not be started.+addClientToServer+  :: IdType -> SiteId -> ClientPort -> Thread+  -> ServerData -> ServerData+addClientToServer i sid cp tid sd+  = sd { sd_ClientMap = nsm', sd_SiteToClientMap = snm', sd_SiteMap = sm' }+  where+    --update the ClientMap+    nsm = sd_ClientMap sd+    nsm' = Map.insert i (ClientInfo i sid cp tid maxLifeValue) nsm+    --update the SiteToClientMap+    snm = sd_SiteToClientMap sd+    snm' = MMap.insert sid i snm+    -- update the SiteMap+    sm = sd_SiteMap sd+    sm' = addIdToMap sid sm+    ++-- | Deletes a new client from the server datastructures,+--   the ping-thread will not be closed.+deleteClientFromServer :: IdType -> ServerData -> ServerData+deleteClientFromServer i sd +  = sd { sd_ClientMap = nsm', sd_SiteToClientMap = snm', sd_SiteMap = sm' }+  where+    --update the ClientMap+    nsm = sd_ClientMap sd+    nsm' = Map.delete i nsm+    --update the SiteToClientMap and the SiteIdMap+    info = lookupClientInfo i sd+    snm = sd_SiteToClientMap sd+    sm = sd_SiteMap sd+    (snm', sm') = deleteSiteId info+    deleteSiteId Nothing = (snm, sm)+    deleteSiteId (Just info') = (MMap.deleteElem sid i snm , deleteIdFromMap sid sm)+      where+      sid = ci_Site info'+++-- | Gets the ClientPort from a ClientId (on the ServerData).+lookupClientInfo :: IdType -> ServerData -> Maybe ClientInfo+lookupClientInfo i sd = Map.lookup i (sd_ClientMap sd)+++-- | Gets a list with all registered clients (on the ServerData).+lookupAllClientInfos :: ServerData -> [ClientInfo]+lookupAllClientInfos sd = Map.elems (sd_ClientMap sd)+++-- | Gets the ClientPort from a ClientId (on the Server).+getClientInfo :: IdType -> Server -> IO (Maybe ClientInfo)+getClientInfo i (Server server)+  = withMVar server $ \sd -> return $ lookupClientInfo i sd +++-- | Gets a list with all registered clients (on the Server).+getAllClientInfos :: Server -> IO [ClientInfo]+getAllClientInfos (Server server)+  = withMVar server $ \sd -> return $ lookupAllClientInfos sd++-- | Sets the life value of a specific client.+setClientLife :: Int -> IdType -> Server -> IO ()+setClientLife v i (Server server)+  = modifyMVar server $+      \sd -> do+        let mbCi = lookupClientInfo i sd+        sd' <- case mbCi of+          (Just ci) -> do+            let ci'  = ci {ci_LifeValue = v}+                nsm  = sd_ClientMap sd+                nsm' = Map.insert i ci' nsm+                sd'  = sd {sd_ClientMap = nsm'}+            return sd'+          (Nothing) -> return sd+        return (sd', ())++-- | Gets the life value of a specific client.+getClientLife :: IdType -> Server -> IO (Int)+getClientLife i (Server server)+  = withMVar server $+      \sd -> do+        let mbCi = lookupClientInfo i sd+        case mbCi of+          (Just ci) -> return $ ci_LifeValue ci+          (Nothing) -> return 0+++instance ServerClass Server where+  registerClient sid po s@(Server server)+    = do+      let cp = newClientPort po+      -- register the client at the server+      (ptid,i,register) <- modifyMVar server $+        \sd ->+        do+        -- create a new Id and a new Port+        let (i, sd') = getNextId sd+        let register = sd_Register sd+        -- add node to controller+        ptid <- newThread+        let sd'' = addClientToServer i sid cp ptid sd'+        return (sd'', (ptid, i, register))+      -- do general registration action+      register i cp+      -- startPingProcess for Client+      setThreadDelay 5000000 ptid+      setThreadAction (checkClient i cp s ptid) ptid+      setThreadErrorHandler (handleCheckClientError i s ptid) ptid+      startThread ptid+      return i+     +  unregisterClient i (Server server)+    = do+      debugM localLogger "unregisterClient: start"+      (mbInfo, unregister) <- modifyMVar server $+        \sd ->+        do+        let unregister = sd_Unregister sd+        let mbInfo = lookupClientInfo i sd +        let sd' = deleteClientFromServer i sd+        return (sd', (mbInfo,unregister))+      debugM localLogger "unregisterClient: client deleted"+      case mbInfo of+        (Just info) -> +          do+          debugM localLogger "unregisterClient: killing the ping thread"+          -- kill the ping-thread+          stopThread (ci_PingThreadId info)+          debugM localLogger "unregisterClient: ping thread killed"+          debugM localLogger "unregisterClient: executing unregister-function"+          -- execute the unregister function+          unregister i (ci_Port info)+          debugM localLogger "unregisterClient: unregister-function executed"+        (Nothing)   ->+          do+          debugM localLogger "unregisterClient: no client info found"+          return ()+      debugM localLogger "unregisterClient: end"+        +  pingServer i (Server server)+    = withMVar server $+        \sd -> return $ isJust $ lookupClientInfo i sd+        +++instance Debug Server where+  printDebug (Server server)+    = withMVar server $+        \sd ->+        do+        putStrLn "printServer"+        putStrLn $ "OwnStream:       " ++ show (sd_OwnStream sd)+        putStrLn $ "OwnPort:         " ++ show (sd_OwnPort sd)+        putStrLn $ "ClientMap:       " ++ show (sd_ClientMap sd)+        putStrLn $ "SiteToClientMap: " ++ show (sd_SiteToClientMap sd)+        putStrLn $ "SiteMap:         " ++ show (sd_SiteMap sd)+        putStrLn $ "NextId:          " ++ show (sd_NextId sd)+  +++-- ----------------------------------------------------------------------------+-- Server-Port+-- ----------------------------------------------------------------------------++-- | The ServerPort is only a wrapper for a Port-Datatype.  +data ServerPort = ServerPort (Port ServerRequestMessage)+  deriving (Show)++instance Binary ServerPort where+  put (ServerPort p) = put p+  get+    = do+      p <- get+      return (ServerPort p)+      ++-- | Creates a new ServerPort.+newServerPort :: StreamName -> Maybe SocketId -> IO ServerPort+newServerPort sn soid+  = do+    p <- newPort sn soid+    return (ServerPort p)++ +instance ServerClass ServerPort where+  registerClient sid po (ServerPort p) +    = do+      withStream $+        \s -> performPortAction p s time30 (SReqRegisterClient sid po) $+          \rsp ->+          do+          case rsp of+            (SRspRegisterClient n) -> return (Just n)+            _ -> return Nothing+  +  unregisterClient i (ServerPort p)+    = do+      withStream $+        \s -> performPortAction p s time30 (SReqUnregisterClient i) $+          \rsp ->+          do+          case rsp of+            (SRspUnregisterClient) -> return (Just ())+            _ -> return Nothing+  +  pingServer i (ServerPort p)+    = do+      withStream $+        \s -> performPortAction p s time30 (SReqPing i) $+          \rsp ->+          do+          case rsp of+            (SRspPing b) -> return (Just b)+            _ -> return Nothing +  +  +++-- ----------------------------------------------------------------------------+-- Client-Messages+-- ----------------------------------------------------------------------------+  +-- | Requests datatype, which is send to a filesystem node.+data ClientRequestMessage+  = CReqPing IdType+  | CReqClientAction B.ByteString+  | CReqClientId+  | CReqServerPort+  | CReqUnknown+  deriving (Show)++instance Binary ClientRequestMessage where+  put (CReqPing i)         = putWord8 1 >> put i+  put (CReqClientAction b) = putWord8 2 >> put b+  put (CReqClientId)       = putWord8 3+  put (CReqServerPort)     = putWord8 4+  put (CReqUnknown)        = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \i -> return (CReqPing i)+        2 -> get >>= \b -> return (CReqClientAction b)+        3 -> return (CReqClientId)+        4 -> return (CReqServerPort)+        _ -> return (CReqUnknown)+++-- | Response datatype from a filesystem node.+data ClientResponseMessage+  = CRspPing Bool+  | CRspClientAction B.ByteString+  | CRspClientId (Maybe IdType)+  | CRspServerPort ServerPort+  | CRspError String+  | CRspUnknown+  deriving (Show)      ++instance Binary ClientResponseMessage where+  put (CRspPing b)         = putWord8 1 >> put b+  put (CRspClientAction b) = putWord8 2 >> put b+  put (CRspClientId i)     = putWord8 3 >> put i+  put (CRspServerPort p)   = putWord8 4 >> put p+  put (CRspError e)        = putWord8 5 >> put e+  put (CRspUnknown)        = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \b -> return (CRspPing b)+        2 -> get >>= \b -> return (CRspClientAction b)+        3 -> get >>= \i -> return (CRspClientId i)+        4 -> get >>= \p -> return (CRspServerPort p)+        5 -> get >>= \e -> return (CRspError e)+        _ -> return (CRspUnknown)++instance RspMsg ClientResponseMessage where+  isError (CRspError _) = True+  isError _ = False+  +  getErrorMsg (CRspError e) = e+  getErrorMsg _ = ""+  +  isUnknown (CRspUnknown) = True+  isUnknown _ = False+  +  mkErrorMsg e = CRspError e++++-- ----------------------------------------------------------------------------+-- Client-TypeClass+-- ----------------------------------------------------------------------------++-- | The request-functions a client has to implement.+class ClientClass c where++  -- | Check, if the client is responding.+  pingClient :: IdType-> c -> IO Bool+  +  -- | Get the ID of the client.+  getClientId :: c -> IO (Maybe IdType)+  +  -- | Gets the server port the client wants to connect to.+  getServerPort :: c -> IO (ServerPort)+        +++  +-- ----------------------------------------------------------------------------+-- Client-Data+-- ----------------------------------------------------------------------------++  +-- | Client datatype.+data ClientData = ClientData {+    cd_ServerThreadId  :: Thread+  , cd_PingThreadId    :: Thread+  , cd_Id              :: Maybe IdType+  , cd_LifeValue       :: Int+  , cd_SiteId          :: SiteId+  , cd_OwnStream       :: Stream ClientRequestMessage+  , cd_OwnPort         :: Port ClientRequestMessage+  , cd_ServerPort      :: ServerPort+  }++  +-- | Only a wrapper around an MVar.+data Client = Client (MVar ClientData)+++-- | Creates a new client, it needs the StreamName and optional +--   the SocketId of the server.+newClient+  :: (Binary a, Binary b)+  => StreamName -> Maybe SocketId+  -> (a -> IO (Maybe b))  -- ^ the individual request dispatcher for the client+  -> IO Client+newClient sn soid action+  = do  +    sp <- newServerPort sn soid+    -- initialize values+    sid     <- getSiteId+    stid    <- newThread  +    st      <- (newLocalStream Nothing::IO (Stream ClientRequestMessage))+    po      <- newPortFromStream st+    ptid    <- newThread+    let cd = (ClientData stid ptid Nothing maxLifeValue sid st po sp)+    c <- newMVar cd+    let client = Client c +    -- first, we start the server, because we can't handle requests without it+    startRequestDispatcher stid st (dispatchClientRequest client action)+    -- then we try to register a the server+    setThreadDelay 5000000 ptid+    setThreadAction (checkServer sp sid po client) ptid+    setThreadErrorHandler (handleCheckServerError client) ptid+    startThread ptid+    return client+++-- | Closes the client.+closeClient :: Client -> IO ()+closeClient (Client client)+  = modifyMVar client $+      \cd ->+      do+      case (cd_Id cd) of+        (Just i)  -> unregisterClient i (cd_ServerPort cd)+        (Nothing) -> return ()+      stopRequestDispatcher (cd_ServerThreadId cd)+      closeStream (cd_OwnStream cd)+      stopThread (cd_PingThreadId cd)+      return (cd, ())+++-- | Handles the requests from the server.+dispatchClientRequest+  :: (Binary a, Binary b)+  => Client+  -> (a -> IO (Maybe b))+  -> ClientRequestMessage +  -> Port ClientResponseMessage+  -> IO ()+dispatchClientRequest client action msg replyPort+  = do+    debugM localLogger $ "dispatchClientRequest: " ++ show msg+    case msg of+      (CReqPing i) ->+        do+        handleRequest replyPort (pingClient i client) (\b -> CRspPing b)+        return ()+      (CReqClientAction b) ->+        do+        -- now, we have a specific client request+        handleRequest replyPort +          (action $ decode b) +          (\res -> maybe (CRspUnknown) (\r -> CRspClientAction $ encode r) res)+      _ -> +        handleRequest replyPort (return ()) (\_ -> CRspUnknown)+++-- | Test, if the client is registered by a server. +isClientRegistered :: Client -> IO (Bool, Maybe IdType)+isClientRegistered (Client client)+  = withMVar client $ \cd -> return $ (isJust (cd_Id cd), (cd_Id cd))+++-- | Deletes the internal clientId, the client will then be in an +--   unregisterd state.+unsetClientId :: Client -> IO ()+unsetClientId (Client client)+  = modifyMVar client $ \cd -> return ( cd {cd_Id = Nothing}, ())++-- | Sets the life value of the server port.+setLifeValue :: Int -> Client -> IO ()+setLifeValue v (Client client)+  = modifyMVar client $ \cd -> return (cd {cd_LifeValue = v}, ())++-- | Gets the life value of the server port.+getLifeValue :: Client -> IO (Int)+getLifeValue (Client client)+  = withMVar client $ \cd -> return (cd_LifeValue cd)++-- | Assigns a new clientId to the client.+setClientId :: IdType -> Client -> IO ()+setClientId i (Client client)+      = modifyMVar client $ \cd -> return ( cd {cd_Id = Just i}, ())  +++instance ClientClass Client where+  pingClient i (Client client)+    = withMVar client $+        \cd -> return $ (cd_Id cd) == (Just i)+        +  getClientId (Client client)+    = withMVar client $+        \cd -> return $ (cd_Id cd)+    +  getServerPort (Client client)+    = withMVar client $+        \cd -> return $ (cd_ServerPort cd)+++instance Debug Client where+  printDebug (Client client)+    = withMVar client $+        \cd ->+        do+        putStrLn "printClient"+        putStrLn $ "Id:         " ++ show (cd_Id cd)+        putStrLn $ "LifeValue   " ++ show (cd_LifeValue cd)+        putStrLn $ "Site:       " ++ show (cd_SiteId cd)+        putStrLn $ "OwnStream:  " ++ show (cd_OwnStream cd)+        putStrLn $ "OwnPort:    " ++ show (cd_OwnPort cd)+        putStrLn $ "ServerPort: " ++ show (cd_ServerPort cd)+++++-- ----------------------------------------------------------------------------+-- Client-Port+-- ----------------------------------------------------------------------------++-- | Just a wrapper around a port.+data ClientPort = ClientPort (Port ClientRequestMessage)+  deriving (Show)++instance Binary ClientPort where+  put (ClientPort p) = put p+  get+    = do+      p <- get+      return (ClientPort p)+++-- | Creates a new ClientPort.+newClientPort :: Port ClientRequestMessage -> ClientPort+newClientPort po = ClientPort po++++instance ClientClass ClientPort where+  pingClient i (ClientPort p)+    = do+      withStream $+        \s -> performPortAction p s time30 (CReqPing i) $+          \rsp ->+          do+          case rsp of+            (CRspPing b) -> return (Just b)+            _ -> return Nothing++  getClientId (ClientPort p)+    = do+      withStream $+        \s -> performPortAction p s time30 (CReqClientId) $+          \rsp ->+          do+          case rsp of+            (CRspClientId i) -> return (Just i)+            _ -> return Nothing+    +  getServerPort (ClientPort p)+    = do+      withStream $+        \s -> performPortAction p s time30 (CReqServerPort) $+          \rsp ->+          do+          case rsp of+            (CRspServerPort sp) -> return (Just sp)+            _ -> return Nothing+++++-- ----------------------------------------------------------------------------+-- Communication-Functions+-- ----------------------------------------------------------------------------++-- | Sends a request from the server to the client an handles the response or+--   invokes a user-defined handler.+sendRequestToClient +  :: (Show a, Binary a, Binary b)+  => ClientPort -> Int+  -> a+  -> (b -> IO (Maybe c))  -- ^ response handler+  -> IO c+sendRequestToClient (ClientPort p) timeout a handler+  = do+    debugM localLogger $ "sending request to Client: " ++ show a+    withStream $+      \s -> performPortAction p s timeout (CReqClientAction (encode a)) $+        \rsp ->+        do+        case rsp of+          (CRspClientAction b) -> +            do+            handler (decode b)+          _ -> return Nothing+++-- | Sends a request from the client to the server an handles the response or+--   invokes a user-defined handler.+sendRequestToServer +  :: (Show a, Binary a, Binary b)+  => ServerPort -> Int+  -> a+  -> (b -> IO (Maybe c))  -- ^ response handler+  -> IO c+sendRequestToServer (ServerPort p) timeout a handler+  = do+    debugM localLogger $ "sending message to Server: " ++ show a+    withStream $+      \s -> performPortAction p s timeout (SReqServerAction (encode a)) $+        \rsp ->+        do+        case rsp of+          (SRspServerAction b) -> +            do+            handler (decode b)+          _ -> return Nothing++          +-- ----------------------------------------------------------------------------+-- Ping-Functions+-- ----------------------------------------------------------------------------++-- | Checks, if a client is still reachable, otherwise it will be deleted+--   from the server.+checkClient :: IdType -> ClientPort -> Server -> Thread -> IO ()+checkClient i po server thread+  = do+    debugM localLogger "pingClient"+    b <- pingClient i po+    case b of+      False -> +        do+        warningM localLogger "pingCient: client is not reachable... delete him"+        handleCheckClientError i server thread+      True  -> +        do+        debugM localLogger "pingCient: client is ok"+        setClientLife maxLifeValue i server+        return ()+    ++-- | If a client does not respond to a ping, this function is invoked.+--   It deleted the client from the server and stops the ping thread.+handleCheckClientError :: IdType -> Server -> Thread -> IO ()+handleCheckClientError i server thread+   = do+     v <- getClientLife i server+     if (v > 0)+       then do+         setClientLife (v-1) i server+       else do+         unregisterClient i server+         stopThread thread+     return ()+++-- | Checks, if a server is still reachable, otherwise it will be +--  deleted from the client.+checkServer :: ServerPort -> SiteId -> Port ClientRequestMessage -> Client -> IO ()+checkServer sepo sid clpo c+  = do+    debugM localLogger "pingServer"+    (reg,i) <- isClientRegistered c+    if (reg)+      then do+        debugM localLogger "pingServer: client is registered, testing server" +        b <- pingServer (fromJust i) sepo+        if (b)+          then do+            debugM localLogger "pingServer: server is ok"+            setLifeValue maxLifeValue c+            return ()              +          else do+            warningM localLogger "pingServer: server is down" +            handleCheckServerError c+      else do +        debugM localLogger "pingServer: trying to register client"+        i' <- registerClient sid clpo sepo+        setClientId i' c    +++-- | If the server does not respond to a ping, this function is invoked.+--   It sets the server to an unregistered state, so it will reconnect again.+handleCheckServerError :: Client -> IO ()+handleCheckServerError c+  = do+    v <- getLifeValue c+    if (v > 0)+      then do setLifeValue (v-1) c+      else do unsetClientId c
+ source/Holumbus/Network/Core.hs view
@@ -0,0 +1,266 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Network.Core+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  The Server-Module for the Holumbus framework.+  +  It contains the lowlevel functions, like the socket handling (opening, +  reading, writing, ...).+  +-}++-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Network.Core+    (+      -- * Socket-Descriptor+      SocketId(..)++      -- * Server-Operations+    , startSocket++      -- * Client-Operations+    , sendRequest++      -- * Handle-Operations+    , putMessage+    , getMessage++    , ThreadIdException(..)+    )+where++import           Prelude hiding         ( catch )++import           Control.Concurrent+import           Control.Exception      ( Exception+					, bracket+					, catch+					)+import           Control.Monad++import           Data.Binary+import qualified Data.ByteString.Lazy   as B+import           Data.Typeable++import           Network+import qualified Network.Socket         as Socket++import           System.Log.Logger+import           System.CPUTime+import           System.IO+import           System.Posix++import           Text.Printf+import           Text.XML.HXT.Arrow++import           Holumbus.Common.Utils  ( handleAll )++-- | Logger+localLogger :: String+localLogger = "Holumbus.Network.Core"+++type ServerDispatcher = SocketId -> Handle -> HostName -> PortNumber -> IO ()++-- ------------------------------------------------------------+--+-- exception stuff++data ThreadIdException  = ThreadIdException ThreadId+                          deriving (Typeable, Show)++instance Exception ThreadIdException where++-- ----------------------------------------------------------------------------+-- Socket Descriptor+-- ----------------------------------------------------------------------------++-- | All data, that is needed to address a socket.+--   Contains the hostname and the portNumber.+data SocketId = SocketId HostName PortNumber +  deriving (Show, Eq)++instance Binary (SocketId) where+  put (SocketId hn po)+    = put hn >> (put . toInteger) po+  get+    = do+      hn <- get+      poInt <- get+      return (SocketId hn (fromInteger poInt))++instance XmlPickler SocketId where+  xpickle = xpSocketId++xpSocketId :: PU SocketId+xpSocketId+  = xpElem "socket" $+    xpWrap(\(hn, po) -> SocketId hn (fromInteger po), \(SocketId hn po) -> (hn, (toInteger po))) $+    xpPair (xpAttr "hostname" xpText) (xpAttr "port" xpickle)+++-- ----------------------------------------------------------------------------+-- Server-Operations+-- ----------------------------------------------------------------------------++-- | Creates a new (unix-)socket and starts the listener in its own thread.+--   You'll get the threadId of the listener Thread, so you can kill it.+--   It is also possible to give a range of PortNumbers on which the socket+--   will be opened. The first portnumber available will be taken.+startSocket +  :: ServerDispatcher -- ^ dispatcher function+  -> PortNumber       -- ^ start port number+  -> PortNumber       -- ^ end port number+  -> IO (Maybe (ThreadId, HostName, PortNumber))+startSocket f actPo maxPo+  = do+    s <- (getFirstSocket actPo maxPo)+    case s of+      Nothing -> +        return Nothing+      (Just (so, po)) ->+        do+        hn <- getHostName+        tid <- forkIO $ +          do+          handleAll+            (\e ->+              do+              putStrLn $ "ERROR - socket closed with exception: " ++ show e +              sClose so+            ) $+            do+            {- 6.8+            catchDyn (waitForRequests f so (SocketId hn po)) (handler so)+            -}+            catch+              (waitForRequests f so (SocketId hn po))+              (handler so)+        return (Just (tid, hn, po))+    where+    handler :: Socket -> ThreadIdException -> IO ()+    handler so (ThreadIdException i)+        = do+          sClose so+          putStrLn $ "socket normally closed by thread " ++ show i +++-- | Gets the hostname of the computer of just "localhost".+getHostName :: IO (HostName)+getHostName+  = do+    (hn, _) <- Socket.getNameInfo [] True False (Socket.SockAddrUnix "localhost")+    return (maybe "localhost" id hn)+++-- | Gets the first free port number and creates of new socket on it.+getFirstSocket :: PortNumber -> PortNumber -> IO (Maybe (Socket, PortNumber))+getFirstSocket actPo maxPo+  | actPo > maxPo = return Nothing+  | otherwise +    = do+      handleAll (return (getFirstSocket (actPo+1) maxPo)) $+        do+        socket <- getSocket (PortNumber actPo)+        return (Just (socket, actPo))     +++-- | Opens a socket on a port number.+getSocket :: PortID -> IO (Socket)+getSocket po =+  -- for MS-Windows Systems+  withSocketsDo $ do+  -- Don't let the server be terminated by sockets closed unexpectedly by the client.+  installHandler sigPIPE Ignore Nothing+  socket <- listenOn po+  return socket+++-- | Listens to a socket and opens a new dispatcher thread for every incomming+--   data.+waitForRequests :: ServerDispatcher -> Socket -> SocketId -> IO ()+waitForRequests f socket soid = +  do+  client <- accept socket+  forkIO $ processRequest f soid client   -- Spawn new thread to answer the current request.+  waitForRequests f socket soid       -- Wait for more requests.+++-- | A wrapper around the user defined dispatcher function.+--   Mesures the time and catches unhandled exceptions.+processRequest :: ServerDispatcher -> SocketId -> (Handle, HostName, PortNumber) -> IO ()+processRequest f soid client = +  bracket (return client) (\(hdl, _, _) -> hClose hdl) (\cl -> processRequest' cl)+    where+    processRequest' (hdl, hst, prt) = +      do+      hSetBuffering hdl NoBuffering+      -- Dispatch the request and measure the processing time.+      t1 <- getCPUTime+      debugM localLogger "starting to dispatch request"+      handleAll (\e -> errorM localLogger $ "UnknownError: " ++ show e) $ do+        f soid hdl hst prt+      t2 <- getCPUTime+      d <- return ((fromIntegral (t2 - t1) / 1000000000000) :: Float)+      ds <- return (printf "%.4f" d)+      infoM localLogger ("request processed in " ++ ds ++ " sec")+++-- ----------------------------------------------------------------------------+-- Client-Operations+-- ----------------------------------------------------------------------------++    +-- | Send the query to a server and merge the result with the global result.+sendRequest :: (Handle -> IO a) -> HostName -> PortID -> IO a+sendRequest f n p = +  withSocketsDo $ do +    installHandler sigPIPE Ignore Nothing+    +    --TODO exception handling+    --handle (\e -> do putStrLn $ show e return False) $+    bracket (connectTo n p) (hClose) (send)+    where    +    send hdl +      = do+        hSetBuffering hdl NoBuffering+        f hdl+++-- ----------------------------------------------------------------------------+-- Handle-Operations+-- ----------------------------------------------------------------------------++-- | Puts a bytestring to a handle. But to make the reading easier, we write+--   the length of the data as a message-header to the handle, too. +putMessage :: B.ByteString -> Handle -> IO ()+putMessage msg hdl+  = do+    handleAll (\e -> do+      errorM localLogger $ "putMessage: " ++ show e+      errorM localLogger $ "message: " ++ show msg +     ) $ do+      hPutStrLn hdl ((show $ B.length msg) ++ " ")+      B.hPut hdl msg+++-- | Reads data from a stream. We define, that the first line of the message+--   is the message header which tells us how much bytes we have to read.+getMessage :: Handle -> IO (B.ByteString)+getMessage hdl+  = do+    line <- hGetLine hdl+    let pkg = words line+    raw <- B.hGet hdl (read $ head pkg)+    return raw+
+ source/Holumbus/Network/Messages.hs view
@@ -0,0 +1,264 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Network.Messages+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  +  General functions for the transmission of messages over the Holumbus-Ports.+  You don't need these functions, but they make your life much easier.+  +  The mailbox concept doesn't deal with the request and response scheme+  very well, but with this module, all the boring stuff is solved. The function+  performPortAction will do everything for you. If you find it boring to write+  a seperate listener-thread for every mailbox you want to read from, you+  might look at the startRequestDispatcher and stopRequestDispatcher functions. ++-}++-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Network.Messages+(+-- * Message-Class+  RspMsg(..)++-- * Port-Handling+, performPortAction++-- * Stream-Handling+, startRequestDispatcher+, stopRequestDispatcher+, handleRequest+)+where++import           Control.Concurrent+import		 Control.Exception	( Exception+					, throw+					)+{- 6.8+import qualified Control.Exception as E+-}++import           Data.Binary+import           Data.Maybe+import           Data.Typeable++import           System.Log.Logger++import           Holumbus.Common.Threading+import           Holumbus.Common.Utils+import qualified Holumbus.Network.Port as P+++localLogger :: String+localLogger = "Holumbus.Network.Messages"++++-- ----------------------------------------------------------------------------+-- Message-Class+-- ----------------------------------------------------------------------------++-- | The typeclass for the response messages. We want to react if an error+--   message is received an this interface helps us to detect and create such+--   a message. The unknown message will be send back, if the server doesn't+--   understand our request.+class RspMsg m where+  isError :: m -> Bool  +  getErrorMsg :: m -> String+  isUnknown :: m -> Bool+  mkErrorMsg :: String -> m++++-- ----------------------------------------------------------------------------+-- Port-Handling+-- ----------------------------------------------------------------------------++-- | Every request might raise an exception+data MessageException+    = TimeoutException     	-- ^ if the server takes too long to respond+    | UnknownRequest       	-- ^ if the server doesn't understand our request+    | FalseResponse String 	-- ^ when the server response doesn't match our definition+    | ErrorResponse String	-- ^ if an error in the server occurred and he informs us about the error+      deriving (Show, Typeable)++instance Exception MessageException++-- | Sends a repest to the server  (stream) and waits for a response.+--   If the response can't be received in a certain time, a TimeoutException+--   will be raised. If a response is received, an individual response handler+--   is executed.+talkWithServer+  :: (Show a, Binary a, Show b, Binary b, RspMsg b)+  => P.Port a          -- ^ port to which the message will be send+  -> P.Stream b        -- ^ the stream from which the response is read+  -> Int               -- ^ timeout for the response in nanoseconds (1000000 = 1 sec) (0 = wait for ever)+  -> a                 -- ^ message to be send+  -> (b -> IO c)       -- ^ handler function for the response +  -> IO c+talkWithServer p respStream timeout m hdlFct+  = do+    respPort <- P.newPortFromStream respStream+    -- send the request to the node+    debugM localLogger $ "sending: " ++ show m+    P.sendWithGeneric p m (encode respPort)+    --wait for the response+    debugM localLogger $ "waiting for response for: " ++ show m +    response <- P.tryWaitReadStream respStream timeout+    -- r' <- P.readStream respStream+    -- let response = Just r'+    debugM localLogger "response Message..."+    debugM localLogger $ show response+    res <- case response of+      -- if no response+      Nothing -> do+        warningM localLogger "talkWithServer: timeout"+        {- E.throwDyn TimeoutException -}+	throw TimeoutException+      -- handle the response+      (Just r) ->+        hdlFct r+    return res+       ++-- | A wrapper around the user defined response handler.+--   All error and unkown response will be catched, so you don't have to+--   deal with them. But you can't also throw an error in your response +--   function, if you want.+basicResponseHandler+  :: (Show b, Binary b, RspMsg b) +  => (b -> IO (Maybe c)) +  -> b +  -> IO c+basicResponseHandler hdlFct rsp+  = do+    -- look for right message+    res <- hdlFct rsp+    case res of+      -- if right type... return result+      (Just r) -> +        return r+      -- else handle error types+      Nothing ->+        handleError+    where+      handleError+        | (isError rsp)   = do+			    warningM localLogger $ "basicResponseHandler: error: " ++ show rsp+			    {- 6.8 E.throwDyn $ ErrorResponse $ getErrorMsg rsp -}+			    throw $ ErrorResponse $ getErrorMsg rsp+        | (isUnknown rsp) = do+			    warningM localLogger $ "basicResponseHandler: unknown: " ++ show rsp+			    {- 6.8 E.throwDyn UnknownRequest -}+			    throw UnknownRequest+        | otherwise       = do+			    warningM localLogger $ "basicResponseHandler: false: " ++ show rsp+			    {- E.throwDyn $ FalseResponse $ show rsp -}+			    throw $ FalseResponse $ show rsp+        +                    +-- | Sends a request to the server (stream) and handles the response and all+--   error cases. Very helpful when simulating a request response scheme+--   with the mailboxes.                    +performPortAction+  :: (Show a, Binary a, Show b, Binary b, RspMsg b) +  => P.Port a             -- ^ request port+  -> P.Stream b           -- ^ response Stream +  -> Int                  -- ^ timeout for the response in mikroseconds (1000000 = 1 sec) (0 = wait for ever)+  -> a                    -- ^ request message+  -> (b -> IO (Maybe c))  -- ^ response handler+  -> IO c+performPortAction reqPo resStr timeout reqMsg rspHdl+  = do+    talkWithServer reqPo resStr timeout reqMsg $+      basicResponseHandler rspHdl+  ++-- ----------------------------------------------------------------------------+-- Stream-Handling+-- ----------------------------------------------------------------------------++-- | The server-side request dispatcher handles all incomming responses.+--   The dispatcher runs in its own thread and should not be killed by+--   any exceptions which will be raised in the handling process.+startRequestDispatcher+  :: (Binary a, Show a, Show b, Binary b, RspMsg b) +  => Thread                                     -- ^ threadId, to be filled+  -> P.Stream a                                 -- ^ request-Stream (this is where the messages come in)+  -> (a -> P.Port b -> IO ())                   -- ^ the dispatcher (create a reply message)+  -> IO ()+startRequestDispatcher thread reqS dispatcher+  = do+    setThreadAction (requestDispatcher reqS dispatcher) thread+    startThread thread+++-- | Stops the request dispatcher.+stopRequestDispatcher :: Thread -> IO ()+stopRequestDispatcher thread+  = do+    stopThread thread +++-- | Wrapper around the user-defined dispatching function. For every incomming+--   request a new thread will be created to be able to handle the next request. +requestDispatcher +  :: (Binary a, Show a, Show b, Binary b, RspMsg b)+  => P.Stream a +  -> (a -> P.Port b -> IO ())+  -> IO ()+requestDispatcher reqS dispatcher+  = do+    -- read the next message from the stream (block, if no message arrived)+    msg <- P.readStreamMsg reqS+    -- extract the data+    let dat = P.getMessageData msg+    debugM localLogger "dispatching new Message... "+    debugM localLogger $ show dat+    -- extract the (possible replyport)+    let responsePort = decodeMaybe $ P.getGenericData msg+    if (isNothing responsePort)+      then do+        warningM localLogger "no reply port in message"+        -- yield+      else do+        -- do the dispatching in a new process...+        _ <- forkIO $ dispatcher dat $ (fromJust responsePort)+        return ()+++-- | Execute a function and send its result to the specified port.+handleRequest+  :: (Show b, Binary b, RspMsg b)+  => P.Port b    -- ^ the reply port (where the messages will be send to)+  -> IO c        -- ^ the action which will generate the data to be send+  -> (c -> b)    -- ^ create an output from the data+  -> IO ()+handleRequest po fhdl fres+  = do+    -- in case, we can't send the error...+    handleAll (\e -> do+        errorM localLogger $ "handleRequest: exeption raised and could not be send to controller" +        errorM localLogger $ show e) $ do+      do+      -- in case our operation fails, we send a failure-response+      handleAll ( \e -> do+		        errorM localLogger $ "handleRequest: exeption raised and reporting to controller" +		        errorM localLogger $ show e+                        P.send po (mkErrorMsg $ show e)+		) $+                do+		-- our action, might raise an exception+		r <- fhdl+		-- send the response+		P.send po $ fres r
+ source/Holumbus/Network/Port.hs view
@@ -0,0 +1,982 @@+-- ----------------------------------------------------------------------------++{- |+  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+
+ source/Holumbus/Network/PortRegistry.hs view
@@ -0,0 +1,81 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Network.PortRegistry+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  The typeclass for the PortRegistry. Contains all PortRegistry functions which+  can accessed remotely by the clients.+  +-}+-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Network.PortRegistry+    (+      -- * Type-Classes+      PortRegistry(..)++      -- * GenericRegistry+    , GenericRegistry+    , mkGenericRegistry+    )+where++import Holumbus.Network.Core++-- ----------------------------------------------------------------------------+-- Type-Class+-- ----------------------------------------------------------------------------++-- | The Interface of the PortRegistry.+class PortRegistry pr where++  -- | Creates a new port entry in the registry.+  registerPort :: String -> SocketId -> pr -> IO ()++  -- | Deletes an entry from the registry.+  unregisterPort :: String -> pr -> IO ()++  -- | Get the socket of the port by its name.+  lookupPort :: String -> pr -> IO (Maybe SocketId)++  -- | Get a list of all registered ports.+  getPorts :: pr -> IO [(String, SocketId)]+  ++++-- ----------------------------------------------------------------------------+-- GenericRegistry+-- ----------------------------------------------------------------------------+  +-- | The generic registry object.+--   This is a wrapper around an PortRegistryData or PortRegistryPort object.+--   With this additional indirection, we eliminate the distinction between+--   the port or the data object in the datatypes using the PortRegistry.+--   Therefore it is easier to get access to the registry and the code gets +--   more readable.+--   This might be a good example of hiding network-access. To the caller it+--   makes no difference, if the PortRegistry is in the same address space of+--   on another computer in the network.++data GenericRegistry = forall r. (PortRegistry r) => GenericRegistry r+++-- | Creates a new generic PortRegistry.+mkGenericRegistry :: (PortRegistry r) => r -> GenericRegistry+mkGenericRegistry = GenericRegistry+++instance PortRegistry GenericRegistry where+  registerPort sn soid (GenericRegistry r) = registerPort sn soid r+  unregisterPort sn (GenericRegistry r) = unregisterPort sn r+  lookupPort sn (GenericRegistry r) = lookupPort sn r+  getPorts (GenericRegistry r) = getPorts r
+ source/Holumbus/Network/PortRegistry/Messages.hs view
@@ -0,0 +1,121 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Network.PortRegistry.Messages+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  This modules defines the messages from and to the PortRegistry.++-}+-- ----------------------------------------------------------------------------++module Holumbus.Network.PortRegistry.Messages+(+  PortRegistryRequestStream+, PortRegistryRequestPort++, PortRegistryResponseStream+, PortRegistryResponsePort++, PortRegistryRequestMessage(..)+, PortRegistryResponseMessage(..)+)+where++import           Data.Binary++import           Holumbus.Network.Port+import           Holumbus.Network.Core+import           Holumbus.Network.Messages+++-- ----------------------------------------------------------------------------+-- Ports+-- ----------------------------------------------------------------------------++-- | The Stream for the messages TO the PortRegistry.+type PortRegistryRequestStream  = Stream PortRegistryRequestMessage++-- | The Port for the messages TO the PortRegistry.+type PortRegistryRequestPort    = Port PortRegistryRequestMessage++-- | The Stream for the messages FROM the PortRegistry.+type PortRegistryResponseStream = Stream PortRegistryResponseMessage++-- | The Port for the messages FROM the PortRegistry.+type PortRegistryResponsePort   = Port PortRegistryResponseMessage++++-- ----------------------------------------------------------------------------+-- Messages+-- ----------------------------------------------------------------------------++-- | The messages TO the PortRegistry.+data PortRegistryRequestMessage+  = PRReqRegister StreamName SocketId+  | PRReqUnregister StreamName+  | PRReqLookup StreamName+  | PRReqGetPorts+  | PRReqUnknown+  deriving (Show)++instance Binary PortRegistryRequestMessage where+  put (PRReqRegister sn soid) = putWord8 1 >> put sn >> put soid+  put (PRReqUnregister sn)    = putWord8 2 >> put sn+  put (PRReqLookup sn)        = putWord8 3 >> put sn+  put (PRReqGetPorts)         = putWord8 4+  put (PRReqUnknown)          = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \sn -> get >>= \soid -> return (PRReqRegister sn soid)+        2 -> get >>= \sn -> return (PRReqUnregister sn)+        3 -> get >>= \sn -> return (PRReqLookup sn)+        4 -> return (PRReqGetPorts)+        _ -> return (PRReqUnknown)+++-- | The messages FROM the PortRegistry.+data PortRegistryResponseMessage+  = PRRspSuccess+  | PRRspLookup (Maybe SocketId)+  | PRRspGetPorts [(String,SocketId)]+  | PRRspError String+  | PRRspUnknown+  deriving (Show)++instance Binary PortRegistryResponseMessage where+  put (PRRspSuccess)     = putWord8 1+  put (PRRspLookup soid) = putWord8 2 >> put soid+  put (PRRspGetPorts ls) = putWord8 3 >> put ls+  put (PRRspError e)     = putWord8 4 >> put e+  put (PRRspUnknown)     = putWord8 0+  get+    = do+      t <- getWord8+      case t of+        1 -> return (PRRspSuccess)+        2 -> get >>= \soid -> return (PRRspLookup soid)+        3 -> get >>= \ls -> return (PRRspGetPorts ls)+        4 -> get >>= \e -> return (PRRspError e)+        _ -> return (PRRspUnknown)++instance RspMsg PortRegistryResponseMessage where+  isError (PRRspError _) = True+  isError _ = False+  +  getErrorMsg (PRRspError e) = e+  getErrorMsg _ = ""+  +  isUnknown (PRRspUnknown) = True+  isUnknown _ = False  +    +  mkErrorMsg e = PRRspError e
+ source/Holumbus/Network/PortRegistry/PortRegistryData.hs view
@@ -0,0 +1,163 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Network.PortRegistry.PortRegistryData+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++  This module contains the main datatype for the PortRegistry.+  +-}+-- ----------------------------------------------------------------------------++module Holumbus.Network.PortRegistry.PortRegistryData+(+-- * Datatypes+  PortRegistryData++-- * Creation and Destruction+, newPortRegistryData+, closePortRegistryData++, getPortRegistryRequestPort++-- * reexport+, setPortRegistry +)+where+++import           Control.Concurrent+import qualified Data.Map as Map+import           Network+import           System.Log.Logger++import           Holumbus.Common.Threading+import           Holumbus.Network.Port+import           Holumbus.Network.Messages+import           Holumbus.Network.PortRegistry+import           Holumbus.Network.PortRegistry.Messages+++localLogger :: String+localLogger = "Holumbus.Network.PortRegistry.PortRegistryData"++++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------++-- | The data needed by the PortRegistry+data PortRegistryData = PortRegistryData {+    prd_ServerThreadId :: Thread                             -- ^ The thread-data of the message-dispatcher thread.+  , prd_OwnStream      :: PortRegistryRequestStream          -- ^ The Stream for all incomming messages.+  , prd_SocketMap      :: MVar (Map.Map StreamName SocketId) -- ^ The map storing the port-data (the real registry).+  }+++-- ----------------------------------------------------------------------------+-- Creation and Destruction+-- ----------------------------------------------------------------------------+++-- | Creates a new PortRegistry.+newPortRegistryData :: StreamName -> Maybe PortNumber -> IO PortRegistryData+newPortRegistryData sn pn+  = do+    st      <- (newStream STLocal (Just sn) pn)::IO PortRegistryRequestStream+    mMVar   <- newMVar Map.empty+    sMVar   <- newThread            +    let prd = PortRegistryData sMVar st mMVar+    startRequestDispatcher sMVar st (dispatch prd)+    return prd+++-- | Closes the PortRegistry with its streams and threads.+closePortRegistryData :: PortRegistryData -> IO ()+closePortRegistryData prd+  = do+    stopRequestDispatcher (prd_ServerThreadId prd)+    closeStream (prd_OwnStream prd)+    return ()+++-- | Get the RequestPort of the PortRegistry.+--   It can be used to give access to the PortRegistry, eg. you can serialize+--   this information and transfer it over the network to grant access to the+--   clients.+getPortRegistryRequestPort :: PortRegistryData -> IO PortRegistryRequestPort+getPortRegistryRequestPort prd = newPortFromStream (prd_OwnStream prd)+++-- | The main dispatch-function. It handles the incomming messages and reacts.+dispatch +  :: PortRegistryData +  -> PortRegistryRequestMessage +  -> PortRegistryResponsePort+  -> IO ()+dispatch prd msg replyPort+  = do+    case msg of+      (PRReqRegister sn soid) ->+        do+        handleRequest replyPort (registerPort sn soid prd) (\_ -> PRRspSuccess)+        return ()+      (PRReqUnregister sn) ->      +        do+        handleRequest replyPort (unregisterPort sn prd) (\_ -> PRRspSuccess)+        return ()+      (PRReqLookup sn) ->+        do+        handleRequest replyPort (lookupPort sn prd) (\soid -> PRRspLookup soid)+        return ()+      (PRReqGetPorts) ->+        do+        handleRequest replyPort (getPorts prd) (\ls -> PRRspGetPorts ls)+        return ()+      _ -> +        do+        infoM localLogger $ "dispatch: unknown request " ++ show msg+        handleRequest replyPort (return ()) (\_ -> PRRspUnknown)++++-- ----------------------------------------------------------------------------+-- Typeclass instanciation (PortRegistry)+-- ----------------------------------------------------------------------------+++-- The PortRegistry-typeclass instanciation for the PortRegistryData.+instance PortRegistry PortRegistryData where+  +  registerPort sn soid prd+    = do+      infoM localLogger $ "register: " ++ sn ++ " - at: " ++ show soid +      modifyMVar (prd_SocketMap prd) $+        \sm -> return (Map.insert sn soid sm, ())++  unregisterPort sn prd+    = do+      infoM localLogger $ "unregister: " ++ sn+      modifyMVar (prd_SocketMap prd) $+        \sm -> return (Map.delete sn sm, ())+  +  lookupPort sn prd+    = do+      infoM localLogger $ "looking up: " ++ sn+      soid <- withMVar (prd_SocketMap prd) $+        \sm -> return (Map.lookup sn sm)+      infoM localLogger $ "result for: " ++ sn ++ " -> " ++ show soid+      return soid+  +  getPorts prd+    = do+      infoM localLogger $ "getting all ports"+      withMVar (prd_SocketMap prd) $+        \sm -> return (Map.toList sm)+
+ source/Holumbus/Network/PortRegistry/PortRegistryPort.hs view
@@ -0,0 +1,144 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Network.PortRegistry.PortRegistryPort+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  +  This module contains the Interface for external access to the PortRegistry.+  +-}+-- ----------------------------------------------------------------------------++module Holumbus.Network.PortRegistry.PortRegistryPort+(+-- * Datatypes+  PortRegistryPort+  +-- * Creation+, newPortRegistryPort+, newPortRegistryFromData+, newPortRegistryFromXmlFile ++-- * reexport+, setPortRegistry +)+where++import System.Log.Logger++import Holumbus.Common.FileHandling+import Holumbus.Network.Port+import Holumbus.Network.Messages+import Holumbus.Network.PortRegistry+import Holumbus.Network.PortRegistry.Messages+++localLogger :: String+localLogger = "Holumbus.Network.PortRegistry.PortRegistryPort"+++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+ +-- | The datatype for the PortRegistry remote interface. +data PortRegistryPort = PortRegistryPort PortRegistryRequestPort+  deriving (Show)+++++-- ----------------------------------------------------------------------------+--  Creation+-- ----------------------------------------------------------------------------+++-- | Creates a new PortRegistryPort from a port-object.+newPortRegistryPort :: PortRegistryRequestPort -> PortRegistryPort+newPortRegistryPort p = PortRegistryPort p+++-- | Creates a new PortRegistryPort from the StreamName and SocketId.+newPortRegistryFromData :: StreamName -> SocketId -> IO PortRegistryPort+newPortRegistryFromData sn soid+  = do+    p <- newPort sn (Just soid)+    return $ newPortRegistryPort p+++-- | Creates a new PortRegistryPort from a XML-File.+newPortRegistryFromXmlFile :: FilePath -> IO PortRegistryPort+newPortRegistryFromXmlFile fp+  = do+    p <- loadFromXmlFile fp+    return $ newPortRegistryPort p++++-- ----------------------------------------------------------------------------+-- Typeclass instanciation for PortRegistryPort+-- ----------------------------------------------------------------------------++-- PortRegistry-typeclass instanciation for the PortRegistryPort.+instance PortRegistry PortRegistryPort where+    +  registerPort sn soid (PortRegistryPort p)+    = do+      debugM localLogger "registerPort start"+      r <- withStream $+        \s -> performPortAction p s time30 (PRReqRegister sn soid) $+          \rsp ->+          do+          case rsp of+            (PRRspSuccess) -> return (Just $ ())+            _ -> return Nothing+      debugM localLogger "registerPort end"+      return r+++  unregisterPort sn (PortRegistryPort p)+    = do+      debugM localLogger "unregisterPort start"+      r <- withStream $+        \s -> performPortAction p s time30 (PRReqUnregister sn) $+          \rsp ->+          do+          case rsp of+            (PRRspSuccess) -> return (Just $ ())+            _ -> return Nothing+      debugM localLogger "unregisterPort end"+      return r+  +  lookupPort sn (PortRegistryPort p)+    = do+      debugM localLogger "lookupPort start"+      r <- withStream $+        \s -> performPortAction p s time30 (PRReqLookup sn) $+          \rsp ->+          do+          case rsp of+            (PRRspLookup soid) -> return (Just $ soid)+            _ -> return Nothing+      debugM localLogger "lookupPort end"+      return r+  ++  getPorts (PortRegistryPort p)+    = do+      debugM localLogger "getPorts start"+      r <- withStream $+        \s -> performPortAction p s time30 (PRReqGetPorts) $+          \rsp ->+          do+          case rsp of+            (PRRspGetPorts ls) -> return (Just $ ls)+            _ -> return Nothing+      debugM localLogger "getPorts end"+      return r+
+ source/Holumbus/Network/Site.hs view
@@ -0,0 +1,219 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Network.Site+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  Just a little Id to help us to decide if two thread are running in the+  same program or just on the same machine or on differen machines. ++-}++-- ----------------------------------------------------------------------------++module Holumbus.Network.Site+(+-- * Datatypes+  SiteId+, SiteMap+  +-- * Operations on the SiteId+, getSiteId+, getSiteHost+, getSiteProcess+, isSameHost+, isSameProcess+, nearestId++-- * Operations on the SiteMap+, emptySiteMap+, addIdToMap+, deleteIdFromMap+, deleteHostFromMap+, isSiteIdMember+, getNeighbourSiteIds+)+where++import           Data.Binary+import qualified Data.List as List+import           Network.Socket+import           System.Posix++import           Text.XML.HXT.Arrow++import qualified Data.Map as Map+import qualified Data.Set as Set+++++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+++-- | The datatype of the SiteId, it contains the hostname and a processid,+--   so it is possible to decide if two site ids belong to the same process+--   or the the same computer or are on distinct computers.+data SiteId = SiteId HostName ProcessID deriving (Show, Eq, Ord)++instance Binary SiteId where+  put (SiteId hn pid) = put hn >> put (toInteger pid)+  get+    = do+      hn <- get+      pid <- get+      return (SiteId hn (fromInteger pid)) ++instance XmlPickler SiteId where+  xpickle = xpSiteId+  +xpSiteId :: PU SiteId+xpSiteId = +  xpElem "siteId" $+      xpWrap (\(hn, pid) -> SiteId hn (fromInteger pid), \(SiteId hn pid) -> (hn, toInteger pid)) xpSite+      where+      xpSite =+        xpPair+          (xpElem "hostname" xpText)+          (xpElem "pid" xpickle)+++-- | Just a little Map to hold the SiteIds an to get the neighbout Ids.+type SiteMap = Map.Map HostName (Set.Set SiteId)+++++-- ----------------------------------------------------------------------------+-- Operations on the SiteId+-- ----------------------------------------------------------------------------+++-- | Little helper function.+getHostName :: IO (HostName)+getHostName+  = do+    (hn, _) <- getNameInfo [] True False (SockAddrUnix "localhost")+    return (maybe "localhost" id hn)+    ++-- | Gets the SiteId for the calling program. +getSiteId :: IO (SiteId)+getSiteId+  = do+    hn <- getHostName+    pid <- getProcessID+    return (SiteId hn pid)++       +-- | Extracts the Hostname from the SiteId.+getSiteHost :: SiteId -> HostName+getSiteHost (SiteId hn _) = hn+++-- | Extracts the ProcessID from the SiteId.+getSiteProcess :: SiteId -> ProcessID+getSiteProcess (SiteId _ pid) = pid+++-- | Test, if the two Ids are located on the same host.+isSameHost :: SiteId -> SiteId -> Bool+isSameHost (SiteId hn1 _) (SiteId hn2 _) = hn1 == hn2+++-- | Test, if the two Ids are located on the same host an in the same process.+isSameProcess :: SiteId -> SiteId -> Bool+isSameProcess = (==)+++-- | The filtering function.+--   Splits the given Id-list into three sublists:+--   the Ids from the same process,+--   the Ids from the same computer,+--   all other Ids.+--   This is some sort of distance function, a simple one, but for now it+--   should do its work.+filterSiteIds :: SiteId -> [SiteId] -> ([SiteId],[SiteId],[SiteId])+filterSiteIds _ [] = ([],[],[])+filterSiteIds i ls+  = (same, local, other)+  where+    (same, temp) = List.partition (\s -> isSameProcess i s) ls+    (local, other) = List.partition (\s -> isSameHost i s) temp+++-- | Gets the nearest item from an Id-list compared to a given Id.+nearestId :: SiteId -> [SiteId] -> Maybe SiteId+nearestId s l = nearestId' $ filterSiteIds s l+  where+  nearestId' ([],  [],  [])  = Nothing+  nearestId' ([],  [],  x:_) = Just x+  nearestId' ([],  x:_, _)   = Just x+  nearestId' (x:_, _,   _)   = Just x+  +  +  +  +-- ----------------------------------------------------------------------------+-- Operations on the SiteMap+-- ----------------------------------------------------------------------------++-- | Empty SiteId-Map.+emptySiteMap :: SiteMap+emptySiteMap = Map.empty+++-- | Adds an id to the map.+addIdToMap :: SiteId -> SiteMap -> SiteMap+addIdToMap i m +  = Map.alter f hn m+    where+      hn = getSiteHost i+      f Nothing = (Just $ Set.singleton i)+      f (Just s) = (Just $ Set.insert i s)+++-- | Deletes an id from the map.+deleteIdFromMap :: SiteId -> SiteMap -> SiteMap+deleteIdFromMap i m+  = Map.alter f hn m+    where+      hn = getSiteHost i+      f Nothing = Nothing+      f (Just s) = filterEmpty $ Set.delete i s+      filterEmpty s+        | s == Set.empty = Nothing+        | otherwise = Just s+++-- | Delete a hostname an all its ids from the map.+deleteHostFromMap :: HostName -> SiteMap -> SiteMap+deleteHostFromMap hn m+  = Map.alter f hn m+    where+      f _ = Nothing+++-- | Test, if the site id is already in the list.+isSiteIdMember :: SiteId -> SiteMap -> Bool +isSiteIdMember i m+  = maybe False (\s -> Set.member i s) (Map.lookup hn m) +    where+      hn = getSiteHost i+++-- | Gets all ids which are on the same host, but not+--   the original siteid itself.+getNeighbourSiteIds :: SiteId -> SiteMap -> Set.Set SiteId+getNeighbourSiteIds i m+  = maybe (Set.empty) (\s -> Set.delete i s) (Map.lookup hn m) +    where+      hn = getSiteHost i