diff --git a/sarsi-hs/Main.hs b/sarsi-hs/Main.hs
--- a/sarsi-hs/Main.hs
+++ b/sarsi-hs/Main.hs
@@ -5,14 +5,14 @@
 import Codec.Sarsi (Event(..), Level(..), Message(..))
 import Codec.Sarsi.GHC (fromGHCLog)
 import Data.Attoparsec.Text.Machine (streamParser)
-import Data.Machine (ProcessT, (<~), asParts, auto, autoM, before, echo, prepended, runT_, source)
-import Data.Machine.Plan (yield)
+import Data.Machine (ProcessT, (<~), asParts, auto, autoM, prepended, runT_, source)
 import Data.List (foldl')
+import Sarsi (getBroker, getTopic)
 import Sarsi.Producer (produce)
 import System.Environment (getArgs)
-import System.IO (stdin, stdout, stderr)
-import System.IO.Machine (byLine, printer)
-import System.Process (CreateProcess, StdStream(..), shell, std_in, std_out, std_err)
+import System.IO (stderr)
+import System.IO.Machine (byLine)
+import System.Process (StdStream(..), shell, std_in, std_err)
 import System.Process.Machine (callProcessMachines, mStdErr)
 import System.Exit (ExitCode, exitWith)
 
@@ -54,5 +54,7 @@
 main :: IO ()
 main = do
   args  <- getArgs
-  ec    <- produce "." $ producer (concat $ List.intersperse " " args)
+  b     <- getBroker
+  t     <- getTopic b "."
+  ec    <- produce t $ producer (concat $ List.intersperse " " args)
   exitWith ec
diff --git a/sarsi-nvim/Data/MessagePack/RPC.hs b/sarsi-nvim/Data/MessagePack/RPC.hs
--- a/sarsi-nvim/Data/MessagePack/RPC.hs
+++ b/sarsi-nvim/Data/MessagePack/RPC.hs
@@ -36,12 +36,13 @@
   tpe <- MP.getInt
   case tpe of
     1 -> do
-      id <- MP.getInt
+      id' <- MP.getInt
       err <- get
       res <- get
-      return $ Response id $ f err res where
+      return $ Response id' $ f err res where
         f ObjectNil res = Success res
         f err ObjectNil = Error err
+        f _ _           = Error ObjectNil
     2 -> do
       method <- MP.getStr
       (ObjectArray params) <- get
diff --git a/sarsi-nvim/Main.hs b/sarsi-nvim/Main.hs
--- a/sarsi-nvim/Main.hs
+++ b/sarsi-nvim/Main.hs
@@ -1,12 +1,12 @@
 module Main where
 
-import Codec.Sarsi (Event(..), Level(..), Location(..), Message(..), getEvent)
-import Data.Machine ((<~), asParts, auto, scan)
+import Codec.Sarsi (Event(..), Level(..), Location(..), Message(..))
+import Data.Machine (ProcessT, (<~), asParts, auto, final, scan, sinkPart_, runT)
 import Data.MessagePack.Object (Object(..), toObject)
 import NVIM.Client (Command(..), runCommand)
-import Sarsi (title)
-import Sarsi.Consumer (consume)
-import System.IO (BufferMode(NoBuffering), hSetBinaryMode, hSetBuffering, stdin, stdout)
+import Sarsi (getBroker, getTopic, title)
+import Sarsi.Consumer (consumeOrWait)
+import System.IO (BufferMode(NoBuffering), hSetBuffering, stdin, stdout)
 import System.IO.Machine (sinkIO)
 
 import qualified Data.Text as Text
@@ -46,11 +46,16 @@
 main = do
   hSetBuffering stdin NoBuffering
   hSetBuffering stdout NoBuffering
-  consume "." $ sinkIO publish <~ asParts <~ auto unpack <~ scan f (True, [])
+  b     <- getBroker
+  t     <- getTopic b "."
+  consumeOrWait t consumer
     where
-      f (first, _) event = convert first event
-      unpack (_, xs) = xs
+      consumer Nothing  src = consumer (Just True) src
+      consumer (Just b) src = do
+        b' <- runT $ final <~ sinkPart_ id (sinkIO publish <~ asParts) <~ converter b <~ src
+        return (Left $ head b')
+      converter :: Bool -> ProcessT IO Event (Bool, [Command])
+      converter b = scan f (b, []) where f (first, _) event = convert first event
       publish cmd = do
-        -- TODO Log failed command in a file (or print to stderr?).
         _ <- runCommand cmd
         return ()
diff --git a/sarsi-nvim/NVIM/Client.hs b/sarsi-nvim/NVIM/Client.hs
--- a/sarsi-nvim/NVIM/Client.hs
+++ b/sarsi-nvim/NVIM/Client.hs
@@ -2,17 +2,15 @@
 
 -- | A primitive NVIM RPC synchronous client.
 
-import Data.Binary.Get (runGet)
 import Data.Binary.Machine (streamGet)
 import Data.Binary.Put (runPut)
 import Data.ByteString (hGetSome, hPut)
-import Data.ByteString.Lazy (fromStrict, toStrict)
+import Data.ByteString.Lazy (toStrict)
 import Data.Machine ((<~), asParts, auto, final, run, source)
 import Data.MessagePack.Object (Object(..), toObject)
 import Data.MessagePack.RPC (Answer(..), Request(..), Message(..), getMessage, putRequest)
 import Data.Text (Text)
 import System.IO (Handle, stdin, stdout)
-
 import qualified Data.Text as Text
 import qualified Data.Vector as Vector
 
diff --git a/sarsi-sbt/Main.hs b/sarsi-sbt/Main.hs
--- a/sarsi-sbt/Main.hs
+++ b/sarsi-sbt/Main.hs
@@ -1,41 +1,45 @@
+{-# LANGUAGE Rank2Types #-}
 module Main where
 
 import Codec.Sarsi (Event)
 import Codec.Sarsi.SBT.Machine (eventProcess)
-import Data.Machine (ProcessT, (<~), asParts, auto, autoM, runT_)
+import Data.Machine (ProcessT, (<~), autoM, runT_)
+import Sarsi (getBroker, getTopic)
 import Sarsi.Producer (produce)
 import System.Environment (getArgs)
 import System.Exit (ExitCode, exitWith)
-import System.Process (CreateProcess, StdStream(..), shell, std_in, std_out, std_err)
-import System.Process.Machine (callProcessMachines, mStdOut)
+import System.Process (StdStream(..), shell, std_in, std_out)
+import System.Process.Machine (ProcessMachines, callProcessMachines)
 import System.IO (BufferMode(NoBuffering), hSetBuffering, stdin, stdout)
-import System.IO.Machine (byChunk, printer)
+import System.IO.Machine (IOSink, byChunk)
 
-import qualified Data.Char as Char
 import qualified Data.List as List
-import qualified Data.Text as Text
 import qualified Data.Text.IO as TextIO
 import qualified Sarsi as Sarsi
 
 title :: String
 title = concat [Sarsi.title, "-sbt"]
 
-callShell :: String -> ProcessT IO Event a -> IO (ExitCode, [a])
-callShell cmd sink = callProcessMachines byChunk createProc (mStdOut pipeline)
-  where
-    pipeline = sink <~ eventProcess <~ echoText stdout
-    echoText h = autoM $ (\txt -> TextIO.hPutStr h txt >> return txt)
-    createProc  = (shell cmd) { std_in = Inherit, std_out = CreatePipe }
+-- TODO Contribute to machines-process
+mStdOut_ :: ProcessT IO a b -> ProcessMachines a a0 k0 -> IO ()
+mStdOut_ mp (_, Just stdOut, _)  = runT_ $ mp <~ stdOut
+mStdOut_ _  _                    = return ()
 
-producer :: String -> ProcessT IO Event Event -> IO ExitCode
+producer :: String -> ProcessT IO Event Event -> IO (ExitCode)
 producer cmd sink = do
-  (ec, _)   <- callShell cmd sink
+  (ec, _) <- callProcessMachines byChunk createProc (mStdOut_ pipeline)
   return ec
+    where
+      pipeline = sink <~ eventProcess <~ echoText stdout
+      echoText h = autoM $ (\txt -> TextIO.hPutStr h txt >> return txt)
+      createProc  = (shell cmd) { std_in = Inherit, std_out = CreatePipe }
 
 main :: IO ()
 main = do
   hSetBuffering stdin NoBuffering
   hSetBuffering stdout NoBuffering
   args  <- getArgs
-  ec    <- produce "." $ producer $ concat $ List.intersperse " " ("sbt":args)
+  b     <- getBroker
+  t     <- getTopic b "."
+  ec    <- produce t $ producer $ concat $ List.intersperse " " ("sbt":args)
   exitWith ec
diff --git a/sarsi.cabal b/sarsi.cabal
--- a/sarsi.cabal
+++ b/sarsi.cabal
@@ -1,5 +1,5 @@
 name:                sarsi
-version:             0.0.1.0
+version:             0.0.2.0
 synopsis:            A universal quickfix toolkit and his protocol.
  
 description:
@@ -34,6 +34,7 @@
     Data.Attoparsec.Text.Machine
   build-depends:
       base                  >= 4.6.0.1    && < 5
+    , async                 >= 2.1        && < 2.2
     , attoparsec            >= 0.12       && < 0.14
     , binary                >= 0.7        && < 0.9
     , bytestring            >= 0.10       && < 0.11
@@ -41,6 +42,7 @@
     , cryptonite            >= 0.10       && < 0.14
     , directory             >= 1.2        && < 1.3
     , filepath              >= 1.4        && < 1.5
+    , fsnotify              >= 0.2        && < 0.3
     , machines              >= 0.6        && < 0.7
     , machines-binary       >= 0.3.0.2    && < 0.4
     , machines-process      >= 0.2.0.6    && < 0.3
@@ -48,6 +50,7 @@
     , msgpack               >= 1.0        && < 1.1
     , network               >= 2.6        && < 2.7
     , process               >= 1.1        && < 1.5
+    , stm                   >= 2.4        && < 2.5
     , text                  >= 1.2        && < 1.3
     , vector                >= 0.10       && < 0.12
 
@@ -58,9 +61,9 @@
   build-depends:       
       base
     , Cabal
-    , sarsi
+    , sarsi                 == 0.0.2.0
   hs-source-dirs:       sarsi
-  ghc-options:          -Wall
+  ghc-options:          -Wall -threaded
   default-language:     Haskell2010
 
 
@@ -68,7 +71,7 @@
   main-is:              Main.hs
   build-depends:       
       base
-    , sarsi                 == 0.0.1.0
+    , sarsi                 == 0.0.2.0
     , machines
     , machines-io
     , machines-process
@@ -76,7 +79,7 @@
     , text
     , fsnotify              >= 0.2        && < 0.3
   hs-source-dirs:       sarsi-hs
-  ghc-options:          -Wall -dynamic
+  ghc-options:          -Wall -dynamic -threaded
   default-language:     Haskell2010
 
 executable sarsi-nvim
@@ -90,7 +93,7 @@
     -- NVIM.Info
   build-depends:       
       base
-    , sarsi                 == 0.0.1.0
+    , sarsi                 == 0.0.2.0
     , machines
     , binary
     , bytestring
@@ -105,7 +108,7 @@
     , vector
     , unordered-containers  >= 0.2        && < 0.3
   hs-source-dirs:       sarsi-nvim
-  ghc-options:          -Wall -dynamic
+  ghc-options:          -Wall -dynamic -threaded
   default-language:     Haskell2010
 
 executable sarsi-sbt
@@ -113,12 +116,12 @@
   other-modules:
   build-depends:       
       base
-    , sarsi                 == 0.0.1.0
+    , sarsi                 == 0.0.2.0
     , machines
     , machines-io
     , machines-process
     , process
     , text
   hs-source-dirs:       sarsi-sbt
-  ghc-options:          -Wall -dynamic
+  ghc-options:          -Wall -dynamic -threaded
   default-language:     Haskell2010
diff --git a/sarsi/Main.hs b/sarsi/Main.hs
--- a/sarsi/Main.hs
+++ b/sarsi/Main.hs
@@ -1,7 +1,7 @@
 module Main where
 
 import Distribution.Text
-import Sarsi (mkSockAddr, title)
+import Sarsi (getBroker, getSockAddr, getTopic, title)
 import Sarsi.Trace (traceHS, traceSBT)
 import System.IO (stdin)
 import System.Environment (getArgs)
@@ -14,5 +14,6 @@
   run ["--trace-sbt"] = traceSBT stdin
   run ["--version"]   = putStrLn $ concat [title, "-", display version]
   run _ = do
-    sock <- mkSockAddr "."
-    print sock
+    b     <- getBroker
+    t     <- getTopic b "."
+    print $ getSockAddr t
diff --git a/src/Codec/Sarsi.hs b/src/Codec/Sarsi.hs
--- a/src/Codec/Sarsi.hs
+++ b/src/Codec/Sarsi.hs
@@ -15,7 +15,7 @@
   | Notify { message :: Message }
 
 instance Show Event where
-  show (Start label) = concat ["starting ", unpack label, " build"]
+  show (Start lbl) = concat ["starting ", unpack lbl, " build"]
   show (Finish 0 0) = "build success"
   show (Finish 0 w) = concat ["build success with ", show w, " warning(s)"]
   show (Finish e 0) = concat ["build fail with ", show e, " error(s)"]
diff --git a/src/Codec/Sarsi/GHC.hs b/src/Codec/Sarsi/GHC.hs
--- a/src/Codec/Sarsi/GHC.hs
+++ b/src/Codec/Sarsi/GHC.hs
@@ -1,7 +1,7 @@
 module Codec.Sarsi.GHC where
 
 import Codec.Sarsi
-import Data.Text (Text, strip)
+import Data.Text (strip)
 
 import qualified Codec.GHC.Log as Log
 import qualified Data.Vector as Vector
diff --git a/src/Codec/Sarsi/SBT.hs b/src/Codec/Sarsi/SBT.hs
--- a/src/Codec/Sarsi/SBT.hs
+++ b/src/Codec/Sarsi/SBT.hs
@@ -33,18 +33,17 @@
   fp  <- takeWhile1 (\c -> c /= sepChar && c /= '\n' && c /= '\r') <* char sepChar
   ln  <- decimal <* char sepChar
   t   <- space *> (untilLineBreak <* "\n")
-  ts  <- manyTill' (lineStart *> (untilLineBreak <* "\n")) (lookAhead $ column)
-  col <- column
+  ts  <- manyTill' (lineStart *> (untilLineBreak <* "\n")) (lookAhead $ column')
+  col <- column'
   _   <- end
   return $ Message (Location fp (col) ln) lvl $ formatTxts t ts
     where
-      takeLineBreak = takeWhile1 $ \w -> w == '\n' || w == '\r'
       level = choice [string "[error]" *> return Error, string "[warn]" *> return Warning]
       lineStart = level <* space
       sepChar = ':'
       formatTxts t [] = Vector.singleton t
       formatTxts t ts = Vector.fromList $ t : init ts
-      column = level *> ((length <$> many1 space) <* "^\n")
+      column' = level *> ((length <$> many1 space) <* "^\n")
 
 cleanEC :: Parser Text
 cleanEC = choice [noEC, withEC]
@@ -62,6 +61,9 @@
     isEscEnd 109 = True
     isEscEnd _ = False
 
+untilLineBreak :: Parser Text
 untilLineBreak = takeWhile1 $ \w -> w /= '\n' && w /= '\r'
+
+end :: Parser ()
 end = choice [const () <$> "\n", endOfInput, return ()]
 
diff --git a/src/Codec/Sarsi/SBT/Machine.hs b/src/Codec/Sarsi/SBT/Machine.hs
--- a/src/Codec/Sarsi/SBT/Machine.hs
+++ b/src/Codec/Sarsi/SBT/Machine.hs
@@ -3,7 +3,7 @@
 import Codec.Sarsi (Event(..), Message(..), Level(..))
 import Codec.Sarsi.SBT (SBTEvent(..), eventParser, cleanEC, untilLineBreak, end)
 import Data.Attoparsec.Text.Machine (processParser, streamParser)
-import Data.Machine (ProcessT, (<~), asParts, auto, autoM, scan, runT_)
+import Data.Machine (ProcessT, (<~), asParts, auto, scan)
 import Data.Text (Text)
 
 import qualified Data.Text as Text
diff --git a/src/Data/Attoparsec/Machine.hs b/src/Data/Attoparsec/Machine.hs
--- a/src/Data/Attoparsec/Machine.hs
+++ b/src/Data/Attoparsec/Machine.hs
@@ -1,8 +1,8 @@
 {-# LANGUAGE Rank2Types #-}
 module Data.Attoparsec.Machine where
 
-import Data.Attoparsec.Internal.Types (IResult(..), Parser)
-import Data.Machine (MachineT(..), ProcessT, Step(Await, Yield), Is(Refl), source, stopped)
+import Data.Attoparsec.Internal.Types (IResult(..))
+import Data.Machine (MachineT(..), ProcessT, Step(Await, Yield), Is(Refl), stopped)
 
 streamParserWith :: (Monoid i, Monad m) => (i -> IResult i a) -> ProcessT m i (Either String a)
 streamParserWith runParser = start where
diff --git a/src/Sarsi.hs b/src/Sarsi.hs
--- a/src/Sarsi.hs
+++ b/src/Sarsi.hs
@@ -3,7 +3,7 @@
 import Crypto.Hash (Digest, hash)
 import Crypto.Hash.Algorithms (MD5)
 import Network.Socket (Family(AF_UNIX), SockAddr(SockAddrUnix), Socket, SocketType(Stream), defaultProtocol, socket)
-import System.Directory (getTemporaryDirectory, doesFileExist, makeAbsolute, removeFile)
+import System.Directory (getTemporaryDirectory, createDirectory, doesDirectoryExist, doesFileExist, makeAbsolute, removeFile)
 import System.FilePath ((</>))
 
 import qualified Data.ByteString.Char8 as BSC8
@@ -11,26 +11,35 @@
 title :: String
 title = "sarsi"
 
--- TODO Windows compat: create a TCP socket.
+newtype Broker = Broker FilePath
 
-mkSocket :: IO Socket
-mkSocket = do
-  socket AF_UNIX Stream defaultProtocol
+data Topic = Topic Broker FilePath
 
-mkSockAddr :: FilePath -> IO SockAddr
-mkSockAddr fp = do
-  path  <- getSockPath fp
-  return . SockAddrUnix $ path
+getBroker :: IO Broker
+getBroker = do
+  tmp <- getTemporaryDirectory
+  let bp = tmp </> title
+  let broker = Broker bp
+  exists <- doesDirectoryExist bp
+  if exists then return broker else createDirectory bp >> return broker
 
-mkSockAddr' :: FilePath -> IO SockAddr
-mkSockAddr' fp = do
-  path  <- getSockPath fp
+getTopic :: Broker -> FilePath -> IO Topic
+getTopic b@(Broker bp) fp' = do
+  fp    <- makeAbsolute fp'
+  return $ Topic b $ bp </> (show $ (hash $ BSC8.pack fp :: Digest MD5))
+
+removeTopic :: Topic -> IO ()
+removeTopic (Topic _ fp) = removeFile fp
+
+createSocket :: IO Socket
+createSocket = do
+  socket AF_UNIX Stream defaultProtocol
+
+createSockAddr :: Topic -> IO SockAddr
+createSockAddr t@(Topic _ path) = do
   exists <- doesFileExist path
   if (exists) then removeFile path else return ()
-  mkSockAddr fp
+  return $ getSockAddr t
 
-getSockPath :: FilePath -> IO FilePath
-getSockPath fp' = do
-  fp    <- makeAbsolute fp'
-  tmp <- getTemporaryDirectory
-  return $ tmp </> concat [title, "-", show $ (hash $ BSC8.pack fp :: Digest MD5)]
+getSockAddr :: Topic -> SockAddr
+getSockAddr (Topic _ path) = SockAddrUnix path
diff --git a/src/Sarsi/Consumer.hs b/src/Sarsi/Consumer.hs
--- a/src/Sarsi/Consumer.hs
+++ b/src/Sarsi/Consumer.hs
@@ -2,29 +2,47 @@
 module Sarsi.Consumer where
 
 import Codec.Sarsi (Event, getEvent)
-import Data.Binary.Machine (streamGet)
-import Data.Machine ((<~), asParts, auto, runT_)
-import Network.Socket (Socket, accept, bind, listen, close, socketToHandle)
-import Sarsi (mkSocket, mkSockAddr')
-import System.IO (IOMode(ReadMode), hClose)
-import System.IO.Machine (IOSink, byChunk, sourceHandle)
-
-consume :: FilePath -> IOSink Event -> IO ()
-consume fp sink = do
-  sock  <- mkSocket
-  addr  <- mkSockAddr' fp
-  bind sock addr
-  listen sock 1
-  serve sock sink
-  close sock
+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
+import Control.Exception (IOException, bracket, try)
+import Data.Binary.Machine (processGet)
+import Data.Machine ((<~), auto, asParts, runT_)
+import Network.Socket (connect, socketToHandle)
+import Sarsi (Broker(..), Topic(..), createSocket, getSockAddr)
+import System.FSNotify (eventPath, watchDir, withManager)
+import System.IO (IOMode(ReadMode), hClose, hWaitForInput)
+import System.IO.Machine (IOSink, IOSource, byChunkOf, sourceHandle)
 
-serve :: Socket -> IOSink Event -> IO ()
-serve sock sink = do
-  (conn, _) <- accept sock
-  h     <- socketToHandle conn ReadMode
-  runT_ $ sink <~ asParts <~ auto unpack <~ streamGet getEvent <~ sourceHandle (byChunk) h
-  hClose h
-  serve sock sink
+consumeOrWait :: Topic -> (Maybe s -> IOSource Event -> IO (Either s a)) -> IO a
+consumeOrWait topic@(Topic (Broker bp) tp) f = do
+  res <- consume topic f
+  either (const $ withManager waitAndRetry) return res
     where
-      unpack (Right e) = [e]
-      unpack (Left _) = []
+      waitAndRetry mng = do
+        lck <- newEmptyMVar
+        stop <- watchDir mng bp pred' $ const $ putMVar lck ()
+        takeMVar lck
+        stop
+        consumeOrWait topic f
+      pred' e = eventPath e == tp
+
+consumeOrWait_ :: Topic -> IOSink Event  -> IO a
+consumeOrWait_ topic@(Topic _ _) sink =
+    consumeOrWait topic f where f _ src = fmap (const $ Left ()) $ runT_ $ sink <~ src
+
+consume :: Topic -> (Maybe s -> IOSource Event -> IO (Either s a)) -> IO (Either IOException a)
+consume topic f = try $ consume' topic f
+
+consume' :: Topic -> (Maybe s -> IOSource Event -> IO (Either s a)) -> IO a
+consume' topic f = bracket createHandle hClose (process Nothing)
+  where
+    createHandle = do
+      sock  <- createSocket
+      connect sock $ getSockAddr topic
+      socketToHandle sock ReadMode
+    process s h = do
+      sa  <- f s $ asParts <~ auto unpack <~ processGet getEvent <~ sourceHandle (byChunkOf 1) h
+      _   <- hWaitForInput h (-1)
+      either (continue h) return sa
+        where continue h' s' = process (Just s') h'
+    unpack (Right e) = [e]
+    unpack (Left _) = []
diff --git a/src/Sarsi/Producer.hs b/src/Sarsi/Producer.hs
--- a/src/Sarsi/Producer.hs
+++ b/src/Sarsi/Producer.hs
@@ -1,26 +1,67 @@
 {-# LANGUAGE Rank2Types #-}
 module Sarsi.Producer where
 
-import Codec.Sarsi (Event(..), putEvent)
+import Codec.Sarsi (Event(Start), putEvent)
+import Control.Exception (IOException, bracket, tryJust)
+import Control.Concurrent.Async (async, cancel, wait)
+import Control.Concurrent.Chan (dupChan, newChan, readChan, writeChan)
+import Control.Concurrent.MVar (modifyMVar_, newMVar, readMVar)
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TQueue (newTQueue, tryReadTQueue, writeTQueue)
 import Data.Binary.Machine (processPut)
-import Data.Machine (ProcessT, (<~), auto, asParts, runT_)
-import Data.Machine.Fanout (fanout)
-import Network.Socket (connect, socketToHandle)
-import System.IO (IOMode(AppendMode), hClose)
-import System.IO.Machine (byChunk, sinkHandle)
-
-import qualified Data.Text as T
-
-import Sarsi (mkSocket, mkSockAddr)
+import Data.Machine (ProcessT, (<~), runT_, sinkPart_, prepended)
+import Network.Socket (Socket, accept, bind, close, listen, socketToHandle)
+import Sarsi (Topic, createSocket, createSockAddr, removeTopic)
+import System.IO (IOMode(WriteMode), Handle, hClose)
+import System.IO.Machine (byChunk, sinkIO, sinkHandle, sourceIO)
 
-produce :: FilePath -> (ProcessT IO Event Event -> IO a) -> IO a
-produce fp f = do
-  sock  <- mkSocket
-  addr  <- mkSockAddr fp
-  connect sock addr
-  h     <- socketToHandle sock AppendMode
-  -- TODO use sinkPart_
-  res   <- f $ asParts <~ fanout [auto (:[]), auto (const []) <~ sinkHandle byChunk h <~ processPut putEvent]
-  hClose h
-  return res
+produce :: Topic -> (ProcessT IO Event Event -> IO a) -> IO a
+produce t f = do
+  conns   <- atomically $ newTQueue
+  chan    <- newChan
+  state   <- newMVar []
+  server  <- async $ bracket bindSock close (serve (process conns chan state))
+  feeder  <- async $ f $ sinkPart_ (\x -> (x, x)) (sinkIO $ feed chan state)
+  a       <- wait feeder
+  waitFinish conns
+  cancel server
+  removeTopic t
+  return a
+    where
+      bindSock = do
+        sock  <- createSocket
+        addr  <- createSockAddr t
+        bind sock addr
+        listen sock 1
+        return sock
+      process conns chan' state h = do
+        chan <- dupChan chan'
+        es   <- readMVar state
+        conn <- async $ do
+          runT_ $ sinkHandle byChunk h <~ processPut putEvent <~ (prepended $ reverse es) <~ (sourceIO $ readChan chan)
+          hClose h
+        atomically $ writeTQueue conns conn
+        return Nothing
+      feed chan state e = do
+        modifyMVar_ state $ case e of
+          (Start _) -> const $ return [e]
+          _         -> return . (:) e
+        writeChan chan e
+      waitFinish conns = do
+        conn <- atomically $ tryReadTQueue conns
+        _    <- tryJust io $ maybe (return ()) wait conn
+        return ()
+          where
+            io :: IOException -> Maybe ()
+            io _ = Just ()
 
+serve :: (Handle -> IO (Maybe a)) -> Socket -> IO a
+serve f sock = bracket acceptHandle hClose process
+  where
+    acceptHandle = do
+      (conn, _) <- accept sock
+      h         <- socketToHandle conn WriteMode
+      return h
+    process h = do
+      a         <- f h
+      maybe (serve f sock) return a