diff --git a/Opts.hs b/Opts.hs
new file mode 100644
--- /dev/null
+++ b/Opts.hs
@@ -0,0 +1,30 @@
+module Opts where
+
+import           Options.Applicative
+
+data WatchOpt = WatchOpt { watchPath   :: String
+                         , includePath :: String  -- ^ an reg exp to include particular files when watching dir
+                         , excludePath :: String  -- ^ an reg exp to exclude particular files when watching dir
+                         , throttlingDelay :: Int     -- ^ milliseconds to wait for duplicate events
+                         , actionCmd   :: [String]
+                         } deriving (Show)
+
+watchOpt :: Parser WatchOpt
+watchOpt = WatchOpt
+     <$> strOption (long "path"
+                    <> metavar "PATH"
+                    <> help "directory / file to watch" )
+     <*> strOption (long "include"
+                    <> value []
+                    <> metavar "INCLUDE"
+                    <> help "pattern for including files")
+     <*> strOption (long "exclude"
+                    <> value []
+                    <> metavar "EXCLUDE"
+                    <> help "pattern for excluding files")
+     <*> option auto (long "throttle"
+                    <> value 0
+                    <> metavar "MILLIS"
+                    <> help "milliseconds to wait for duplicate events")
+     <*> (many . strArgument) (metavar "COMMAND"
+                      <> help "command to run" )
diff --git a/Pipeline.hs b/Pipeline.hs
new file mode 100644
--- /dev/null
+++ b/Pipeline.hs
@@ -0,0 +1,50 @@
+-- poor man's streaming library, using threads communicating with MVars
+module Pipeline where
+
+import Prelude hiding (id, (.))
+
+import Control.Category
+import Control.Concurrent (ThreadId, forkIO, threadDelay)
+import Control.Concurrent.MVar
+import Control.Monad
+
+
+newtype Pipeline a b = Pipeline {
+  -- launches zero or more threads, forming a pipeline which takes from
+  -- the 'MVar a' and puts to the 'MVar b'.
+  runPipeline :: MVar a -> IO ([ThreadId], MVar b)
+}
+
+instance Category Pipeline where
+    id = Pipeline $ \inputMVar -> return ([], inputMVar)
+    s2 . s1 = Pipeline $ \inputMVar -> do
+      (threads1, intermediateMVar) <- runPipeline s1 inputMVar
+      (threads2, outputMVar) <- runPipeline s2 intermediateMVar
+      return (threads1 ++ threads2, outputMVar)
+
+
+mkPipeline :: (a -> IO b) -> Pipeline a b
+mkPipeline f = Pipeline $ \inputMVar -> do
+    outputMVar <- newEmptyMVar
+    threadId <- forkIO $ forever $ do
+      x <- takeMVar inputMVar
+      y <- f x
+      putMVar outputMVar y
+    return ([threadId], outputMVar)
+
+
+-- ignore duplicate events, defined as events closer than 'delay'
+-- milliseconds apart
+throttle :: Int -> Pipeline () ()
+throttle delay = Pipeline $ \inputMVar -> do
+    intermediateMVar <- newEmptyMVar
+    outputMVar <- newEmptyMVar
+    absorberThreadId <- forkIO $ forever $ do
+      () <- takeMVar inputMVar
+      void $ tryPutMVar intermediateMVar () -- don't block if the next thread is sleeping
+    sleepingThreadId <- forkIO $ forever $ do
+      () <- takeMVar intermediateMVar
+      threadDelay (1000 * delay)
+      _ <- tryTakeMVar intermediateMVar -- drop any event received while sleeping
+      putMVar outputMVar ()
+    return ([absorberThreadId, sleepingThreadId], outputMVar)
diff --git a/Setup.hs b/Setup.hs
--- a/Setup.hs
+++ b/Setup.hs
@@ -1,3 +1,2 @@
 import Distribution.Simple
-
 main = defaultMain
diff --git a/fswatcher.cabal b/fswatcher.cabal
--- a/fswatcher.cabal
+++ b/fswatcher.cabal
@@ -1,7 +1,7 @@
 name:               fswatcher
 category:           Tools
 build-type:         Simple
-version:            0.2.0
+version:            0.2.1
 synopsis:           Watch a file/directory and run a command when it's modified
 description:        A simple program that watches a file or a directory and
                     runs a given command whenever the file or a file within the
@@ -27,6 +27,8 @@
                    , regex-pcre-builtin   >= 0.94
     ghc-options:     -Wall
     main-is:         fswatcher.hs
+    other-modules:   Opts
+                   , Pipeline
 
 source-repository head
   type:     git
diff --git a/fswatcher.hs b/fswatcher.hs
--- a/fswatcher.hs
+++ b/fswatcher.hs
@@ -1,5 +1,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 
+import Prelude hiding (id, (.))
+
 import System.IO (hPutStrLn, stderr)
 import System.Posix.Files (getFileStatus, isDirectory)
 import System.Directory (canonicalizePath, getCurrentDirectory)
@@ -10,6 +12,7 @@
        stopManager, watchTree, watchDir, eventPath)
 import System.Exit (ExitCode (..), exitSuccess)
 import System.Process (createProcess, proc, waitForProcess)
+import Control.Category
 import Control.Monad (void)
 import System.Posix.Signals (installHandler, Handler(Catch), sigINT, sigTERM)
 import Control.Concurrent (forkIO, killThread)
@@ -18,6 +21,7 @@
 import Options.Applicative
 
 import Opts
+import Pipeline
 
 data FileType = File | Directory deriving Eq
 
@@ -74,10 +78,17 @@
   s <- getFileStatus canonicalPath
   let filetype = if isDirectory s then Directory else File
 
-  runTrigger <- newEmptyMVar
-  runThread <- forkIO $ runCmd cmd args runTrigger
-  stopWatcher <- watch filetype m canonicalPath runTrigger opt
+  -- Check if throttling was requested.
+  let delay = throttlingDelay opt
+  let pipeline = if delay > 0 then throttle delay else id
 
+  inputMVar <- newEmptyMVar
+  (pipelineThreads, outputMVar) <- runPipeline pipeline inputMVar
+  runThread <- forkIO $ runCmd cmd args outputMVar
+  stopWatcher <- watch filetype m canonicalPath inputMVar opt
+  
+  let allThreads = runThread : pipelineThreads
+
   -- Calculate the full path in order to print the "real" file when watching a
   -- path with one or more symlinks.
   currDir <- getCurrentDirectory
@@ -92,7 +103,7 @@
   putStrLn "\nStopping."
   stopWatcher
   stopManager m
-  killThread runThread
+  mapM_ killThread allThreads
   exitSuccess
 
 main :: IO ()