packages feed

heartbeat-streams (empty) → 0.1.0.0

raw patch · 4 files changed

+157/−0 lines, 4 filesdep +asyncdep +basedep +io-streamssetup-changed

Dependencies added: async, base, io-streams, time

Files

+ LICENSE view
@@ -0,0 +1,29 @@+BSD 3-Clause License++Copyright (c) 2016-present, Bitnomial, Inc.+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 the copyright holder nor the names of its+  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 HOLDER 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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ heartbeat-streams.cabal view
@@ -0,0 +1,31 @@+name:                heartbeat-streams+version:             0.1.0.0+synopsis:            Heartbeats for io-streams+License:             BSD3+license-file:        LICENSE+author:              Matthew Wraith+maintainer:          matt@bitnomial.com, opensource@bitnomial.com+copyright:           Bitnomial, Inc. (c) 2016+category:            IO-Streams, Concurrency+build-type:          Simple+cabal-version:       >=1.10+homepage:            https://github.com/bitnomial/heartbeat-streams+bug-reports:         https://github.com/bitnomial/heartbeat-streams/issues++source-repository head+  type: git+  location: git://github.com/bitnomial/heartbeat-streams.git++library+  hs-source-dirs:   src/lib+  default-language: Haskell2010+  ghc-options:+    -Wall -fwarn-tabs -fno-warn-unused-do-bind+    -funbox-strict-fields -O2++  exposed-modules: System.IO.Streams.Heartbeat++  build-depends: async      >= 2.0 && < 2.2+               , base       >= 4.8 && < 4.11+               , io-streams >= 1.1 && < 1.6+               , time       >= 1.4 && < 1.9
+ src/lib/System/IO/Streams/Heartbeat.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE BangPatterns #-}++module System.IO.Streams.Heartbeat+    ( heartbeatOutputStream+    , heartbeatInputStream+    , HeartbeatException (..)+    ) where++import           Control.Concurrent       (threadDelay)+import           Control.Concurrent.Async (async, cancel, link)+import           Control.Exception        (Exception, throw)+import           Control.Monad            (forever)+import           Data.IORef               (atomicModifyIORef', newIORef, writeIORef)+import           Data.Time.Clock          (DiffTime, UTCTime, diffTimeToPicoseconds, diffUTCTime, getCurrentTime)+import           System.IO.Streams        (InputStream, OutputStream)+import qualified System.IO.Streams        as Streams+++-- | Send a message 'a' if nothing has been written on the stream for some interval of time.+-- Writing 'Nothing' to this 'OutputStream' is required for proper cleanup.+heartbeatOutputStream :: DiffTime -- ^ Heartbeat interval+                      -> a        -- ^ Heartbeat message+                      -> OutputStream a -> IO (OutputStream a)+heartbeatOutputStream interval msg os = do+    t <- newIORef =<< getCurrentTime+    writeAsync <- async $ delayInterval >> forever (writeHeartbeat t)+    link writeAsync+    Streams.makeOutputStream (resetHeartbeat t writeAsync)+  where+    delayInterval = delayDiffTime interval++    writeHeartbeat t = do+        !now <- getCurrentTime+        (!timeTilHeartbeat, !triggerHeartbeat) <- atomicModifyIORef' t (heartbeatTime interval now)++        if triggerHeartbeat+            then Streams.write (Just msg) os >> delayInterval+            else delayDiffTime timeTilHeartbeat++    resetHeartbeat t _ x@(Just _)       = Streams.write x os >> getCurrentTime >>= writeIORef t+    resetHeartbeat _ writeAsync Nothing = Streams.write Nothing os >> cancel writeAsync+++-- | Exception to kill the heartbeat monitoring thread+-- Heartbeat Exceptions carry the grace period, ie. the last time a message was received+data HeartbeatException = MissedHeartbeat DiffTime deriving (Show, Eq)+instance Exception HeartbeatException+++-- | Grace period = grace time multiplier x heartbeat interval+-- Usually something like graceMultiplier = 2 is a good idea.+--+-- This throws a 'MissedHeartbeat' exception if a heartbeat is not+-- received within the grace period.+heartbeatInputStream :: DiffTime -- ^ Heartbeat interval+                     -> DiffTime -- ^ Grace time multiplier+                     -> InputStream a -> IO (InputStream a)+heartbeatInputStream interval graceMultiplier is = do+    t <- newIORef =<< getCurrentTime+    checkAsync <- async $ delayDiffTime gracePeriod >> forever (checkHeartbeat t)+    link checkAsync+    -- If disconnect is received, cancel heartbeat watching thread+    Streams.mapM_ (resetHeartbeat t) is >>= Streams.atEndOfInput (cancel checkAsync)+  where+    gracePeriod = graceMultiplier * interval++    checkHeartbeat t = do+        !now <- getCurrentTime+        !triggerDisconnect <- snd <$> atomicModifyIORef' t (heartbeatTime gracePeriod now)++        if triggerDisconnect+            then throw (MissedHeartbeat gracePeriod)+            else delayDiffTime interval++    resetHeartbeat t _ = getCurrentTime >>= writeIORef t+++-- | This is structured to work nicely with 'atomicModifyIORef'. Given+-- the heartbeat interval and the current timestamp, calculate if a+-- heartbeat must be sent and how much time there is until the next heartbeat+-- must be sent.+heartbeatTime :: DiffTime -- ^ Maximum time since last message, ie. heartbeat interval or grace period+              -> UTCTime  -- ^ Current timestamp+              -> UTCTime  -- ^ Last message timestamp+              -> (UTCTime, (DiffTime, Bool)) -- ^ (New last message timestamp, (time til heartbeat, send new message?))+heartbeatTime interval now lastTime = (if triggerHeartbeat then now else lastTime, (timeTilHeartbeat, triggerHeartbeat))+  where+    timeSinceMsg = realToFrac $ diffUTCTime now lastTime+    triggerHeartbeat = timeSinceMsg >= interval+    timeTilHeartbeat = interval - timeSinceMsg+++delayDiffTime :: DiffTime -> IO ()+delayDiffTime = threadDelay . picosToMicros+  where picosToMicros = fromIntegral . diffTimeToPicoseconds . (/ 1000000)