hworker-ses (empty) → 0.1.0.0
raw patch · 6 files changed
+276/−0 lines, 6 filesdep +aesondep +amazonkadep +amazonka-sessetup-changed
Dependencies added: aeson, amazonka, amazonka-ses, base, hworker, lens, text, time, unordered-containers
Files
- CHANGELOG.md +3/−0
- LICENSE +13/−0
- README.md +74/−0
- Setup.hs +2/−0
- hworker-ses.cabal +32/−0
- src/System/Hworker/SES.hs +152/−0
+ CHANGELOG.md view
@@ -0,0 +1,3 @@+* 0.1.0.0 Daniel Patterson <dbp@dbpmail.net> 2015-10-25++ Initial release.
+ LICENSE view
@@ -0,0 +1,13 @@+Copyright (c) 2015 Daniel Patterson++Permission to use, copy, modify, and/or distribute this software for any purpose+with or without fee is hereby granted, provided that the above copyright notice+and this permission notice appear in all copies.++THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND+FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS+OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER+TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF+THIS SOFTWARE.
+ README.md view
@@ -0,0 +1,74 @@+## About++This is a library that facilitates sending email via AWS SES using the+background processor [hworker](http://hackage.haskell.org/package/hworker). In+particular, it handles rate limiting (for sending rate currently, not+daily quotas), as SES does not queue messages.++## Rate limiting++Aside from sending emails in the background, the main thing that+`hworker-ses` provides is rate limiting. The rate limiting is only on+a per-process (most likely, per server) basis, as it is controlled via+in-memory storage. This means that you should take into account how+many workers you are likely to have when setting the+messages-per-second rate on the workers.++For example, if your rate limit is 30 messages per second, and you expect+to have at most 5 workers (you can organize them however you like, but a+simple strategy is just to start a worker with each application process),+then if you set the rate limit to be 6 messages per second, things will be+fine.++Note that if you do run over, it won't mean that your messages actually+don't get delivered. That error code will trigger a retry on the message,+so eventually it will go out. But it's wasteful (as you can know in advance+if a given message will actually be able to be sent), and if you really+mess it up (ie, set an absurdly high limit), you potentially will have+your API calls rate limited, which isn't good.++## Usage++The program in the `example` directory is probably most of what you+need to get started. In particular, you create an `hworker` (you need+to give it a name for the queue - this should be shared by any servers+that should be accessing the same queue and sending the same messages,+though if you are sharing the redis server with other applications,+this should be distinct). The third argument is the per-second rate+limit for this server, and the last argument is the address that the+messages are sent from.++Using what you get back, you can spawn workers and a monitor. Then you+can queue as many messages as you want, and they will be sent out by+the workers.++You need to start at least one worker thread and at least+one monitor thread. You can have as many of these as you want, though+generally speaking, having more than one monitor thread per server is+probably overkill (for the system to work, you want at least one+monitor running _somewhere_ - so if a whole server crashes, you want+there to be a monitor elsewhere). Having many workers makes things run+faster. In particular, if you have a high sending limit, you will+probably want several workers.++As specified in the [hworker](http://hackage.haskell.org/package/hworker)+documentation, the semantics of this queue is at-least-once, so it's+possible that messages can get sent multiple times in error conditions+(like if an entire server crashes right after the message is sent, but+before the fact that it was sent is acknowledged). But, provided that+redis is still available and is reliable, messages that have been+queued are guaranteed to get sent eventually (even in the case of+servers crashing, etc). The only caveat to that is that for the+message to get sent at least one worker and one monitor thread must be+running (thoses don't need to be running all the time, but as long as+they aren't running, messages might be delayed. Once they're started again,+those delayed messages will go out).++## Primary Libraries Used++- hworker+- amazonka-ses++## Contributors++- Daniel Patterson <dbp@dbpmail.net>
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ hworker-ses.cabal view
@@ -0,0 +1,32 @@+name: hworker-ses+version: 0.1.0.0+synopsis: Library for sending email with Amazon's SES and hworker+description: See README.+homepage: http://github.com/dbp/hworker-ses+license: ISC+license-file: LICENSE+author: Daniel Patterson+maintainer: dbp@dbpmail.net+-- copyright:+category: Web+build-type: Simple+extra-source-files: README.md CHANGELOG.md+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: System.Hworker.SES+ build-depends: base >= 4.7 && < 5+ , hworker+ , time+ , aeson+ , text+ , amazonka+ , amazonka-ses+ , lens+ , unordered-containers+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/dbp/hworker-ses
+ src/System/Hworker/SES.hs view
@@ -0,0 +1,152 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+module System.Hworker.SES ( SESWorker+ , SESWorkerWith+ , SESState+ , SESJob(..)+ , SESConfig(..)+ , RedisConnection(..)+ , defaultSESConfig+ , create+ , createWith+ , queue+ , worker+ , monitor+ , jobs+ ) where++import Control.Applicative ((<|>))+import Control.Arrow ((&&&))+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar)+import Control.Exception (SomeException, catch)+import Control.Lens (set)+import Control.Monad (mzero)+import Control.Monad.Trans.AWS hiding (page)+import Data.Aeson (FromJSON (..), ToJSON (..),+ Value (Object, String), object, (.:),+ (.=))+import qualified Data.HashMap.Strict as HashMap+import Data.Monoid ((<>))+import Data.Text (Text)+import qualified Data.Text as T+import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime)+import GHC.Generics+import Network.AWS.SES hiding (Success)+import System.Hworker hiding (create, createWith)+import qualified System.Hworker as Hworker (create, createWith)++type SESWorkerWith a = Hworker (SESState a) (SESJob a)+type SESWorker = SESWorkerWith ()++data SESState a = SESState { sesLimit :: Int+ , sesSource :: Text+ , sesRecents :: MVar [UTCTime]+ , sesAfter :: SESJob a -> IO ()+ , sesLogger :: forall a. Show a => a -> IO ()+ }++data SESJob a = SESJob { sesEmTo :: Text+ , sesEmSubj :: Text+ , sesEmBodyText :: Maybe Text+ , sesEmBodyHtml :: Maybe Text+ , sesPayload :: Either Value a+ }+ deriving (Generic, Show)+instance ToJSON a => ToJSON (SESJob a) where+ toJSON SESJob{..} = object [ "v" .= (1 :: Int)+ , "t" .= sesEmTo+ , "s" .= sesEmSubj+ , "x" .= sesEmBodyText+ , "h" .= sesEmBodyHtml+ , ("p", either id toJSON sesPayload)+ ]+instance FromJSON a => FromJSON (SESJob a) where+ parseJSON (Object v) = SESJob <$> v .: "t"+ <*> v .: "s"+ <*> v .: "x"+ <*> v .: "h"+ <*> ((Right <$> (v .: "p")) <|>+ pure (Left (HashMap.lookupDefault (String "No 'p' field.") "p" v)))+ parseJSON _ = mzero++instance (ToJSON a, FromJSON a, Show a) => Job (SESState a) (SESJob a) where+ job state@(SESState limit source recents after log) j@(SESJob to subj btxt bhtml payload) =+ do now <- getCurrentTime+ rs <- takeMVar recents+ let active = filter ((< 1) . diffUTCTime now) rs+ let count = length active+ if count >= limit+ then putMVar recents active >> threadDelay 100000 >> job state j+ else do putMVar recents (now : active)+ awsenv <- getEnv NorthVirginia Discover+ r <- runAWST awsenv $+ send (sendEmail source+ (set dToAddresses [to]+ destination)+ (message (content subj)+ (set bHtml+ (content <$> bhtml) $+ set bText+ (content <$> btxt)+ body)))+ case r of+ Left err ->+ do log err+ return (Failure (T.pack (show err)))+ Right _ -> do catch (after j)+ (\(e::SomeException) ->+ log ("hworker-ses callback raised exception: " <> show e))+ return Success++data SESConfig a =+ SESConfig { sesconfigName :: Text+ , sesconfigLimit :: Int+ , sesconfigSource :: Text+ , sesconfigAfter :: (SESJob a -> IO ())+ , sesconfigLogger :: (forall a. Show a => a -> IO ())+ , sesconfigFailedQueueSize :: Int+ , sesconfigRedisConnectInfo :: RedisConnection+ }++defaultSESConfig :: (ToJSON a, FromJSON a, Show a)+ => Text+ -> Int+ -> Text+ -> (SESJob a -> IO ())+ -> SESConfig a+defaultSESConfig name limit source after =+ let d = defaultHworkerConfig name ()+ in SESConfig name+ limit+ source+ after+ (hwconfigLogger d)+ (hwconfigFailedQueueSize d)+ (hwconfigRedisConnectInfo d)++create :: (ToJSON a, FromJSON a, Show a)+ => Text+ -> Int+ -> Text+ -> (SESJob a -> IO ())+ -> IO (SESWorkerWith a)+create name limit source after =+ createWith (defaultSESConfig name limit source after)++createWith :: (ToJSON a, FromJSON a, Show a)+ => SESConfig a+ -> IO (SESWorkerWith a)+createWith (SESConfig name limit source after logger failed redis) =+ do recents <- newMVar []+ Hworker.createWith+ (defaultHworkerConfig name+ (SESState limit source recents after logger)) {+ hwconfigRedisConnectInfo = redis+ ,hwconfigFailedQueueSize = failed+ ,hwconfigLogger = logger+ }