diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/heartbeat-streams.cabal b/heartbeat-streams.cabal new file mode 100644 --- /dev/null +++ b/heartbeat-streams.cabal @@ -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 diff --git a/src/lib/System/IO/Streams/Heartbeat.hs b/src/lib/System/IO/Streams/Heartbeat.hs new file mode 100644 --- /dev/null +++ b/src/lib/System/IO/Streams/Heartbeat.hs @@ -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)