konsum: options nack and requeuenack

[fd/haskell-amqp-utils.git] / agitprop.hs

diff --git a/agitprop.hs b/agitprop.hs
index 7f52a28561ad36374d17208229419a867a51c306..480307a9e5bcd11df2e0cd62bff2a4cc5c9e9d0a 100644 (file)
--- a/agitprop.hs
+++ b/agitprop.hs
@@ -1,13 +1,18 @@
-{-# LANGUAGE OverloadedStrings #-}
-
+-- generic AMQP publisher

 import Control.Concurrent (threadDelay)
 import qualified Control.Exception as X
 import Control.Monad (forever)
 import qualified Data.ByteString.Lazy.Char8 as BL
 import Control.Concurrent (threadDelay)
 import qualified Control.Exception as X
 import Control.Monad (forever)
 import qualified Data.ByteString.Lazy.Char8 as BL

+import Data.List (isSuffixOf)
+import Data.Maybe

 import qualified Data.Text as T
 import qualified Data.Text as T

+import Data.Time
+import Data.Time.Clock.POSIX

 import Data.Version (showVersion)
 import Data.Word (Word64)
 import Data.Version (showVersion)
 import Data.Word (Word64)

+import Magic

 import Network.AMQP
 import Network.AMQP

+import Network.AMQP.Types

 import Network.AMQP.Utils.Connection
 import Network.AMQP.Utils.Helpers
 import Network.AMQP.Utils.Options
 import Network.AMQP.Utils.Connection
 import Network.AMQP.Utils.Helpers
 import Network.AMQP.Utils.Options


@@ -37,14 +42,7 @@ main = do
       confirmSelect chan False
       addConfirmationListener chan confirmCallback
     else return ()
       confirmSelect chan False
       addConfirmationListener chan confirmCallback
     else return ()

-  let publishOneMsg f = do
-        r <-
-          publishMsg
-            chan
-            (T.pack $ currentExchange args)
-            (T.pack $ rKey args)
-            newMsg {msgBody = f, msgDeliveryMode = Just Persistent}
-        printparam "sent" $ fmap show r
+  let publishOneMsg = publishOneMsg' chan args

   X.catch
     (if isDir
        then do
   X.catch
     (if isDir
        then do


@@ -54,22 +52,25 @@ main = do
              inotify
              [CloseWrite, MoveIn]
              (inputFile args)
              inotify
              [CloseWrite, MoveIn]
              (inputFile args)

-             (handleEvent publishOneMsg)
-         hr $ "watching " ++ (inputFile args)
+             (handleEvent publishOneMsg (suffix args) (inputFile args))
+         hr $ "BEGIN watching " ++ (inputFile args)

          _ <- forever $ threadDelay 1000000
          removeWatch wd
          _ <- forever $ threadDelay 1000000
          removeWatch wd

+         hr $ "END watching " ++ (inputFile args)

        else do
        else do

-         hr $ "sending " ++ (inputFile args)
+         hr $ "BEGIN sending"

          messageFile <- BL.readFile (inputFile args)
          if (lineMode args)
          messageFile <- BL.readFile (inputFile args)
          if (lineMode args)

-           then mapM_ publishOneMsg (BL.lines messageFile)
-           else publishOneMsg messageFile)
+           then mapM_ (publishOneMsg Nothing) (BL.lines messageFile)
+           else publishOneMsg (Just (inputFile args)) messageFile
+         hr "END sending")

     (\exception -> printparam' "exception" $ show (exception :: X.SomeException))
   -- all done. wait and close.
   if (confirm args)
     (\exception -> printparam' "exception" $ show (exception :: X.SomeException))
   -- all done. wait and close.
   if (confirm args)

-    then waitForConfirms chan >>= return . show >> return ()
+    then waitForConfirms chan >>= (printparam' "confirmed") . show

     else return ()
   closeConnection conn
     else return ()
   closeConnection conn

+  hr "connection closed"

 
 -- | The handler for publisher confirms
 confirmCallback :: (Word64, Bool, AckType) -> IO ()
 
 -- | The handler for publisher confirms
 confirmCallback :: (Word64, Bool, AckType) -> IO ()


@@ -83,18 +84,73 @@ confirmCallback (deliveryTag, isAll, ackType) =
      (show ackType))
 
 -- | Hotfolder event handler
      (show ackType))
 
 -- | Hotfolder event handler

-handleEvent :: (BL.ByteString -> IO ()) -> Event -> IO ()
+handleEvent ::
+     (Maybe String -> BL.ByteString -> IO ())
+  -> [String]
+  -> String
+  -> Event
+  -> IO ()

 -- just handle closewrite and movedin events
 -- just handle closewrite and movedin events

-handleEvent f (Closed False (Just x) True) = handleFile f x
-handleEvent f (MovedIn False x _) = handleFile f x
-handleEvent _ _ = return ()
+handleEvent f s p (Closed False (Just x) True) = handleFile f s (p ++ "/" ++ x)
+handleEvent f s p (MovedIn False x _) = handleFile f s (p ++ "/" ++ x)
+handleEvent _ _ _ _ = return ()

 
 -- | Hotfolder file handler
 
 -- | Hotfolder file handler

-handleFile :: (BL.ByteString -> IO ()) -> FilePath -> IO ()
-handleFile _ ('.':_) = return () -- ignore hidden files
-handleFile f x =
+handleFile ::
+     (Maybe String -> BL.ByteString -> IO ()) -> [String] -> FilePath -> IO ()
+handleFile _ _ ('.':_) = return () -- ignore hidden files
+handleFile f s@(_:_) x =
+  if any (flip isSuffixOf x) s
+    then handleFile f [] x
+    else return ()
+handleFile f [] x =

   X.catch
   X.catch

-    (hr ("sending " ++ x) >> BL.readFile x >>= f)
+    (BL.readFile x >>= f (Just x))

     (\exception ->
        printparam' "exception in handleFile" $
        show (exception :: X.SomeException))
     (\exception ->
        printparam' "exception in handleFile" $
        show (exception :: X.SomeException))

+
+-- | Publish one message with our settings
+publishOneMsg' :: Channel -> Args -> Maybe String -> BL.ByteString -> IO ()
+publishOneMsg' c a fn f = do
+  printparam "sending" fn
+  (mtype, mencoding) <-
+    if (magic a) && isJust fn
+      then do
+        m <- magicOpen [MagicMimeType]
+        magicLoadDefault m
+        t <- magicFile m (fromJust fn)
+        magicSetFlags m [MagicMimeEncoding]
+        e <- magicFile m (fromJust fn)
+        return (Just (T.pack t), Just (T.pack e))
+      else return ((contenttype a), (contentencoding a))
+  now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds
+  r <-
+    publishMsg
+      c
+      (T.pack $ currentExchange a)
+      (T.pack $ rKey a)
+      newMsg
+        { msgBody = f
+        , msgDeliveryMode = persistent a
+        , msgTimestamp = Just now
+        , msgID = msgid a
+        , msgType = msgtype a
+        , msgUserID = userid a
+        , msgApplicationID = appid a
+        , msgClusterID = clusterid a
+        , msgContentType = mtype
+        , msgContentEncoding = mencoding
+        , msgReplyTo = replyto a
+        , msgPriority = prio a
+        , msgCorrelationID = corrid a
+        , msgExpiration = msgexp a
+        , msgHeaders = substheader (fnheader a) fn $ msgheader a
+        }
+  printparam "sent" $ fmap show r
+  where
+    substheader ::
+         [String] -> Maybe String -> Maybe FieldTable -> Maybe FieldTable
+    substheader (s:r) (Just fname) old =
+      substheader r (Just fname) (addheader old (s ++ "=" ++ fname))
+    substheader _ _ old = old