Changed Connection event to provide XML.Event streams.

1 files changed, 37 insertions, 46 deletions

diff --git a/Presence/Server.hs b/Presence/Server.hs
index 5f85c262..46b32b81 100644
--- a/Presence/Server.hs
+++ b/Presence/Server.hs
@@ -24,6 +24,7 @@ module Server where
 
 import Data.ByteString (ByteString,hGetNonBlocking)
 import qualified Data.ByteString.Char8 as S -- ( hPutStrLn, hPutStr, pack)
+import Data.Conduit ( Source, Sink, Flush )
 #if MIN_VERSION_containers(0,5,0)
 import qualified Data.Map.Strict as Map
 import           Data.Map.Strict (Map)
@@ -160,9 +161,7 @@ data InOrOut = In | Out
 -- | These events may be read from 'serverEvent' TChannel.
 --
 data ConnectionEvent b
-    = Got b
-    -- ^ Arrival of data from a socket
-    | Connection (STM Bool) (IO (Maybe ByteString)) (ByteString -> IO Bool)
+    = Connection (STM Bool) (Source IO b) (Sink (Flush b) IO ())
     -- ^ A new connection was established
     | ConnectFailure SockAddr
     -- ^ A 'Connect' command failed.
@@ -192,16 +191,16 @@ data ConnectionRecord u
 
 -- | This object accepts commands and signals events and maintains
 -- the list of currently listening ports and established connections.
-data Server a u releaseKey
+data Server a u releaseKey b
  = Server { serverCommand    :: TMVar (ServerInstruction a u)
-          , serverEvent      :: TChan ((a,u), ConnectionEvent ByteString)
+          , serverEvent      :: TChan ((a,u), ConnectionEvent b)
           , serverReleaseKey :: releaseKey
           , conmap           :: TVar (Map a (ConnectionRecord u))
           , listenmap        :: TVar (Map PortNumber ServerHandle)
           , retrymap         :: TVar (Map SockAddr (TVar Bool,InterruptibleDelay))
           }
 
-control :: Server a u releaseKey -> ServerInstruction a u -> IO ()
+control :: Server a u releaseKey b -> ServerInstruction a u -> IO ()
 control sv = atomically . putTMVar (serverCommand sv)
 
 type Allocate releaseKey m = forall b. IO b -> (b -> IO ()) -> m (releaseKey, b)
@@ -226,8 +225,12 @@ noCleanUp io _ = ( (,) () ) `liftM` liftIO io
 -- >   let loop = do
 -- >          (_,event) <- atomically $ readTChan (serverEvent sv)
 -- >          case event of
--- >            Got bytes -> putStrLn $ "got: " ++ show bytes
--- >            _         -> return ()
+-- >            Connection getPingFlag readData writeData -> do
+-- >              forkIO $ do
+-- >                fix $ \readLoop -> do
+-- >                    readData >>= mapM $ \bytes ->
+-- >                        putStrLn $ "got: " ++ show bytes
+-- >                        readLoop
 -- >          case event of EOF -> return ()
 -- >                        _   -> loop
 -- >   liftIO loop
@@ -238,8 +241,10 @@ noCleanUp io _ = ( (,) () ) `liftM` liftIO io
 server ::
     -- forall (m :: * -> *) a u conkey releaseKey.
     (Show conkey, MonadIO m, Ord conkey) =>
-    Allocate releaseKey m -> m (Server conkey u releaseKey)
-server allocate = do
+    Allocate releaseKey m
+    -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+    -> m (Server conkey u releaseKey x)
+server allocate sessionConduits = do
     (key,cmds) <- allocate (atomically newEmptyTMVar)
                            (atomically . flip putTMVar Quit)
     server <- liftIO . atomically $ do
@@ -288,6 +293,8 @@ server allocate = do
                  `fmap` atomically (readTVar $ listenmap server)
     when (not listening) $ do
 
+    hPutStrLn stderr $ "Started listening on "++show port
+
     let family  = AF_INET6
         address = case family of
             AF_INET  -> SockAddrInet port iNADDR_ANY
@@ -297,13 +304,14 @@ server allocate = do
         { serverWarn    = hPutStrLn stderr
         , serverSession = \sock _ h -> do
             (conkey,u) <- makeConnKey params sock
-            _ <- newConnection server params conkey u h In
+            _ <- newConnection server sessionConduits params conkey u h In
             return ()
         }
 
     atomically $ listenmap server `modifyTVar'` Map.insert port sserv
 
   doit server (Ignore port) = do
+    hPutStrLn stderr $ "Stopping listen on "++show port
     mb <- atomically $ do
         map <- readTVar $ listenmap server
         modifyTVar' (listenmap server) $ Map.delete port
@@ -342,7 +350,7 @@ server allocate = do
         connect sock addr
         (conkey,u) <- makeConnKey params (restrictSocket sock)
         h <- socketToHandle sock ReadWriteMode
-        newConnection server params conkey u h Out
+        newConnection server sessionConduits params conkey u h Out
         return ()
 
   doit server (ConnectWithEndlessRetry addr params interval) = do
@@ -380,7 +388,7 @@ server allocate = do
             connect sock addr
             (conkey,u) <- makeConnKey params (restrictSocket sock)
             h <- socketToHandle sock ReadWriteMode
-            threads <- newConnection server params conkey u h Out
+            threads <- newConnection server sessionConduits params conkey u h Out
             atomically $ do threadsWait threads
                             readTVar retryVar
         fin_utc <- getCurrentTime
@@ -428,30 +436,29 @@ socketFamily (SockAddrInet6 _ _ _ _) = AF_INET6
 socketFamily (SockAddrUnix _)        = AF_UNIX
 
 
-conevent :: ConnectionState -> ConnectionEvent b
-conevent con = Connection pingflag read write
+conevent :: ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+            -> ConnectionState
+            -> ConnectionEvent x
+conevent sessionConduits con = Connection pingflag read write
  where
-    pingflag = swapTVar (pingFlag (connPingTimer con)) False
-    read =  connRead con
-    write = connWrite con
+    pingflag     = swapTVar (pingFlag (connPingTimer con)) False
+    (read,write) = sessionConduits (connRead con) (connWrite con)
 
 newConnection :: Ord a
-                 => Server a u1 releaseKey
+                 => Server a u1 releaseKey x
+                 -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
                  -> ConnectionParameters conkey u
                  -> a
                  -> u1
                  -> Handle
                  -> InOrOut
                  -> IO ConnectionThreads
-newConnection server params conkey u h inout = do
+newConnection server sessionConduits params conkey u h inout = do
     hSetBuffering h NoBuffering
-    let (forward,idle_ms,timeout_ms) =
+    let (idle_ms,timeout_ms) =
             case (inout,duplex params) of
-                (Out,False) -> ( const $ return ()
-                              , 0
-                              , 0 )
-                _ -> ( announce . ((conkey,u),) . Got
-                     , pingInterval params
+                (Out,False) -> ( 0, 0 )
+                _ -> ( pingInterval params
                      , timeout params )
 
     new <- do pinglogic <- forkPingMachine idle_ms timeout_ms
@@ -473,7 +480,7 @@ newConnection server params conkey u h inout = do
             (newCon,e) <- return $
               if duplex params
                 then let newcon = SaneConnection new
-                     in ( newcon, ((conkey,u), conevent newcon) )
+                     in ( newcon, ((conkey,u), conevent sessionConduits newcon) )
                 else ( case inout of
                         In  -> ReadOnlyConnection new
                         Out -> WriteOnlyConnection new
@@ -496,22 +503,6 @@ newConnection server params conkey u h inout = do
                 kont <- updateConMap conkey u new what
                 putTMVar started ()
                 return kont
-#ifdef TEST
-    -- enable this for 'Got' events
-    forkIO $ do -- inout==In || duplex params then forkIO $ do
-        warn $ "waiting gots thread: " <> bshow (conkey,inout)
-        atomically $ takeTMVar started
-        -- pingBump pinglogic -- start the ping timer
-        fix $ \loop -> do
-        -- warn $ "read thread: " <> bshow (conkey,inout)
-        mb <- threadsRead new
-        -- pingBump pinglogic
-        warn $ "got: " <> bshow (mb,(conkey,inout))
-        maybe (return ())
-              (atomically . forward >=> const loop)
-              mb
-        warn $ "quit-gots: " <> bshow (conkey,inout)
-#endif
     return new
  where
 
@@ -562,17 +553,17 @@ newConnection server params conkey u h inout = do
             announce ((conkey,u),EOF)
             connClose replaced
             let newcon = SaneConnection new
-            announce $ ((conkey,u),conevent newcon)
+            announce $ ((conkey,u),conevent sessionConduits newcon)
             return $ newcon
         else
          case replaced of
             WriteOnlyConnection w | inout==In ->
                  do let newcon = ConnectionPair new w
-                    announce ((conkey,u),conevent newcon)
+                    announce ((conkey,u),conevent sessionConduits newcon)
                     return newcon
             ReadOnlyConnection r  | inout==Out ->
                  do let newcon = ConnectionPair r new
-                    announce ((conkey,u),conevent newcon)
+                    announce ((conkey,u),conevent sessionConduits newcon)
                     return newcon
             _ -> do -- connFlush todo
                     announce ((conkey,u0), EOF)