sarsi 0.0.1.0 → 0.0.2.0

raw patch · 15 files changed

+202/−118 lines, 15 filesdep +asyncdep +stmdep ~sarsiPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async, stm

Dependency ranges changed: sarsi

API changes (from Hackage documentation)

Files

• sarsi-hs/Main.hs +8/−6
• sarsi-nvim/Data/MessagePack/RPC.hs +3/−2
• sarsi-nvim/Main.hs +14/−9
• sarsi-nvim/NVIM/Client.hs +1/−3
• sarsi-sbt/Main.hs +19/−15
• sarsi.cabal +12/−9
• sarsi/Main.hs +4/−3
• src/Codec/Sarsi.hs +1/−1
• src/Codec/Sarsi/GHC.hs +1/−1
• src/Codec/Sarsi/SBT.hs +6/−4
• src/Codec/Sarsi/SBT/Machine.hs +1/−1
• src/Data/Attoparsec/Machine.hs +2/−2
• src/Sarsi.hs +27/−18
• src/Sarsi/Consumer.hs +42/−24
• src/Sarsi/Producer.hs +61/−20

@@ -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

@@ -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

@@ -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 ()

@@ -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 

@@ -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

@@ -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

@@ -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

@@ -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)"]

@@ -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

@@ -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 ()] 

@@ -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

@@ -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

@@ -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

@@ -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 _) = []

@@ -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