avro-piper (empty) → 1.0.1
raw patch · 12 files changed
+354/−0 lines, 12 filesdep +aesondep +avrodep +avro-pipersetup-changed
Dependencies added: aeson, avro, avro-piper, base, bytestring, conduit, conduit-combinators, conduit-extra, hedgehog, hspec, hw-hspec-hedgehog, hw-kafka-avro, mtl, optparse-applicative, scientific, stringsearch, text, unordered-containers
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- README.md +43/−0
- Setup.hs +2/−0
- app/Main.hs +58/−0
- app/Options.hs +35/−0
- avro-piper.cabal +97/−0
- src/Conduit/Splitter.hs +33/−0
- src/Error.hs +18/−0
- src/Format.hs +12/−0
- test/Spec.hs +1/−0
- test/SplitterSpec.hs +22/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for avro-piper++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Alexey Raga (c) 2018++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Alexey Raga nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,43 @@+# avro-piper++A Schema-registry aware avro decoding tool++### Example+This tool is intended to be used with `kafkacat` as:++```+$ kafkacat -q -b 127.0.0.1 -t data-topic -p 4 -o beginning -c 5 -D '---' | avro-decode -r http://127.0.0.1:8081 -D '---' | jq .+{+ "id": 185,+ "name": "Utah Education Network",+ "timestamp": 1473674981000+}+{+ "id": 185,+ "name": "Utah Education Network",+ "timestamp": 1473626112000+}+{+ "id": 185,+ "name": "Utah Education Network",+ "timestamp": 1473628416000+}+{+ "id": 185,+ "name": "Utah Education Network",+ "timestamp": 1473697795000+}+{+ "id": 185,+ "name": "Utah Education Network",+ "timestamp": 1473634687000+}+```++## To build (or not to build)++```+git clone git@github.com:haskell-works/avro-piper.git+cd avro-piper+stack install+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ app/Main.hs view
@@ -0,0 +1,58 @@+module Main where++import Conduit as C+import Control.Arrow (left)+import Control.Monad+import Control.Monad.Except+import qualified Data.Avro as A+import qualified Data.Avro.Decode as A+import qualified Data.Avro.Schema as A+import qualified Data.Avro.Types as A+import qualified Data.ByteString as StrictBS+import qualified Data.ByteString.Char8 as StrictC8++import Data.ByteString.Builder as BB+import Data.ByteString.Lazy (ByteString, fromStrict, null,+ toStrict)+import qualified Data.ByteString.Lazy as BS+import Data.ByteString.Lazy.Search (split)+import Data.Monoid+import Kafka.Avro++import Conduit.Splitter+import Data.Aeson as J++import Error+import Options+++main :: IO ()+main = do+ opt <- parseOptions+ sr <- schemaRegistry (optSchemaRegistry opt)+ runConduit $+ C.stdinC+ .| mapC fromStrict+ .| splitDelim (StrictC8.pack $ optDelimiter opt)+ .| mapMC (\x -> do+ res <- decodeMessage sr x+ case res of+ Left err -> liftIO (print x) >> return res+ _ -> return res+ )++ .| mapC failBadly+ .| mapC J.encode++ .| unlinesAsciiC+ .| mapC toStrict+ .| stdoutC+++decodeMessage :: MonadIO m => SchemaRegistry -> ByteString -> m (Either DecodeError (A.Value A.Type))+decodeMessage sr bs = runExceptT $ do+ (sid, payload) <- asExceptTPure id . maybeToEither BadPayloadNoSchemaId $ extractSchemaId bs+ sch <- asExceptT DecodeRegistryError (loadSchema sr sid)+ asExceptT (DecodeError sch) (pure $ A.decodeAvro sch payload)++
+ app/Options.hs view
@@ -0,0 +1,35 @@+module Options+where++import Data.Semigroup ((<>))+import Format+import Options.Applicative++data Options = Options+ { optSchemaRegistry :: String+ , optDelimiter :: String+ } deriving (Show)++options :: Parser Options+options = Options+ <$> strOption+ ( long "schema-registry"+ <> short 'r'+ <> metavar "URL"+ <> help "Schema registry address (http://localhost:8081)"+ )+ <*> (unescape <$> strOption+ ( long "delim"+ <> short 'D'+ <> metavar "STRING"+ <> help "Delimiter to separate messages on input"+ <> showDefault <> value "\n"+ ))++optionsParser :: ParserInfo Options+optionsParser = info (helper <*> options)+ ( fullDesc+ )++parseOptions :: IO Options+parseOptions = execParser optionsParser
+ avro-piper.cabal view
@@ -0,0 +1,97 @@+cabal-version: 2.4++name: avro-piper+version: 1.0.1+synopsis: Tool for decoding avro+description: Please see the README on Github at <https://github.com/haskell-works/avro-piper#readme>+category: Data+homepage: https://github.com/haskell-works/avro-piper#readme+bug-reports: https://github.com/haskell-works/avro-piper/issues+author: Alexey Raga+maintainer: alexey.raga@gmail.com+copyright: Alexey Raga+license: BSD-3-Clause+license-file: LICENSE+build-type: Simple++extra-source-files: ChangeLog.md+ README.md++source-repository head+ type: git+ location: https://github.com/haskell-works/avro-piper++library+ hs-source-dirs: src+ build-depends: aeson+ , avro+ , base >=4.7 && <5+ , bytestring+ , conduit+ , conduit-combinators+ , conduit-extra+ , hw-kafka-avro+ , mtl+ , optparse-applicative+ , scientific+ , stringsearch+ , text+ , unordered-containers+ exposed-modules: Conduit.Splitter+ Error+ Format+ other-modules: Paths_avro_piper+ autogen-modules: Paths_avro_piper+ default-language: Haskell2010++executable avro-decode+ main-is: Main.hs+ hs-source-dirs: app+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: aeson+ , avro+ , avro-piper+ , base >=4.7 && <5+ , bytestring+ , conduit+ , conduit-combinators+ , conduit-extra+ , hw-kafka-avro+ , mtl+ , optparse-applicative+ , scientific+ , stringsearch+ , text+ , unordered-containers+ other-modules: Options+ Paths_avro_piper+ autogen-modules: Paths_avro_piper+ default-language: Haskell2010++test-suite avro-piper-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs: test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: aeson+ , avro+ , avro-piper+ , base >=4.7 && <5+ , bytestring+ , conduit+ , conduit-combinators+ , conduit-extra+ , hedgehog+ , hspec+ , hw-hspec-hedgehog+ , hw-kafka-avro+ , mtl+ , optparse-applicative+ , scientific+ , stringsearch+ , text+ , unordered-containers+ other-modules: SplitterSpec+ Paths_avro_piper+ autogen-modules: Paths_avro_piper+ default-language: Haskell2010
+ src/Conduit/Splitter.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE LambdaCase #-}+module Conduit.Splitter+where++import Conduit+import Control.Monad+import qualified Data.ByteString as StrictBS+import Data.ByteString.Builder as BB+import Data.ByteString.Lazy (ByteString, fromStrict, null,+ toStrict)+import qualified Data.ByteString.Lazy as BS+import Data.ByteString.Lazy.Search (split)+import Data.Foldable+import Data.Monoid++splitDelim :: Monad m => StrictBS.ByteString -> Conduit ByteString m ByteString+splitDelim delim = go mempty+ where+ go bldr =+ await >>= \case+ Nothing -> yieldB bldr+ Just bs ->+ case split delim bs of+ [] -> yieldB bldr >> go mempty+ [x] -> go (bldr <> BB.lazyByteString x)+ (x:bs) -> do+ let (xs, mbLast) = foldl' (\(as, l) x -> (maybe as (:as) l, Just x)) ([], Nothing) bs+ let lastB = maybe mempty BB.lazyByteString mbLast+ yieldB (bldr <> BB.lazyByteString x)+ forM_ xs yieldNonEmpty+ go lastB+ yieldNonEmpty bs = if not $ BS.null bs then yield bs else pure ()+ yieldB b = yieldNonEmpty $ BB.toLazyByteString b
+ src/Error.hs view
@@ -0,0 +1,18 @@+module Error+where++import Control.Arrow (left)+import Control.Monad.Except++asExceptT :: Monad m => (e -> e') -> m (Either e a) -> ExceptT e' m a+asExceptT f me = ExceptT $ left f <$> me++asExceptTPure :: Monad m => (e -> e') -> Either e a -> ExceptT e' m a+asExceptTPure f e = ExceptT. pure $ left f e++maybeToEither :: e -> Maybe a -> Either e a+maybeToEither e = maybe (Left e) Right++failBadly :: Show e => Either e a -> a+failBadly (Left e) = error (show e)+failBadly (Right a) = a
+ src/Format.hs view
@@ -0,0 +1,12 @@+module Format+where++unescape :: String -> String+unescape = foldr step ""+ where+ step '\\' (c:cs) = case c of+ 'n' -> '\n':cs+ 'r' -> '\r':cs+ 't' -> '\t':cs+ _ -> c:cs+ step c cs = c:cs
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ test/SplitterSpec.hs view
@@ -0,0 +1,22 @@+{-# LANGUAGE OverloadedStrings #-}+module SplitterSpec+where++import Conduit+import Conduit.Splitter+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as LBS+import Data.Conduit.List+import Data.Monoid+import Test.Hspec++{-# ANN module ("HLint: ignore Redundant do" :: String) #-}++spec :: Spec+spec = describe "SplitterSpec" $ do+ let delim = "\n---\n"+ it "should split" $ do+ let res = runConduitPure $ sourceList ["abra", "cadabra" <> delim, "more" <> delim, "good", "ness"]+ .| splitDelim (LBS.toStrict delim)+ .| sinkList+ res `shouldBe` ["abracadabra", "more", "goodness"]