diff --git a/default.nix b/default.nix
--- a/default.nix
+++ b/default.nix
@@ -5,7 +5,7 @@
 
 mkDerivation {
   pname = "kafka-device";
-  version = "1.0.0.0";
+  version = "1.0.2.0";
   src = ./.;
   isLibrary = true;
   isExecutable = true;
diff --git a/kafka-device.cabal b/kafka-device.cabal
--- a/kafka-device.cabal
+++ b/kafka-device.cabal
@@ -1,5 +1,5 @@
 name         : kafka-device
-version      : 1.0.0.0
+version      : 1.0.2.0
 synopsis     : UI device events via a Kafka message broker
 description  : This package contains functions for passing UI device events to topics on a Kafka message broker \<<https://kafka.apache.org/>\>.  Also see \<<https://hackage.haskell.org/package/kafka-device-joystick/>\>, \<<https://hackage.haskell.org/package/kafka-device-glut/>\>, \<<https://hackage.haskell.org/package/kafka-device-spacenav/>\>, \<<https://hackage.haskell.org/package/kafka-device-leap/>\>.
 license      : MIT
diff --git a/src/Network/UI/Kafka.hs b/src/Network/UI/Kafka.hs
--- a/src/Network/UI/Kafka.hs
+++ b/src/Network/UI/Kafka.hs
@@ -24,9 +24,11 @@
 -- * Consumption
 , ConsumerCallback
 , consumerLoop
+, rawConsumerLoop
 -- * Production
 , ProducerCallback
 , producerLoop
+, rawProducerLoop
 ) where
 
 
@@ -84,18 +86,30 @@
 consumerLoop :: TopicConnection             -- ^ The Kafka topic name and connection information.
              -> ConsumerCallback            -- ^ The consumer callback.
              -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.
-consumerLoop TopicConnection{..} consumer =
+consumerLoop topicConnection =
+  let
+    fromMessage :: Message -> (Sensor, Event)
+    fromMessage message =
+      let
+        (_, _, _, Key (Just (KBytes k)), Value (Just (KBytes v))) = _messageFields message
+      in
+        (unpack k, decode $ fromStrict v)
+  in
+    rawConsumerLoop topicConnection fromMessage
+      . uncurry
+
+
+-- | Consume messages for a Kafka topic.
+rawConsumerLoop :: TopicConnection             -- ^ The Kafka topic name and connection information.
+                -> (Message -> a)              -- ^ The decoder.
+                -> (a -> IO ())                -- ^ The consumer callback.
+                -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.
+rawConsumerLoop TopicConnection{..} fromMessage consumer =
   do
     exitFlag <- newEmptyMVar :: IO (MVar ())
     let
       topic' = fromString topic
       address' = fromString *** fromIntegral $ address
-      fromMessage :: Message -> (Sensor, Event)
-      fromMessage message =
-        let
-          (_, _, _, Key (Just (KBytes k)), Value (Just (KBytes v))) = _messageFields message
-        in
-          (unpack k, decode $ fromStrict v)
       loop offset =
         do
           result <- withAddressHandle address' $ \handle -> fetch' handle =<< fetchRequest offset 0 topic'
@@ -104,7 +118,7 @@
             messages = fromMessage . _tamMessage <$> fetchMessages result
           liftIO
             $ do
-              mapM_ (uncurry consumer) messages
+              mapM_ consumer messages
               threadDelay 100 -- FIXME: Is a thread delay really necessary in order not to miss messages?  Why?
           running <- liftIO $ isEmptyMVar exitFlag
           when running
@@ -128,7 +142,19 @@
              -> Sensor                      -- ^ The name of the sensor producing the event.
              -> ProducerCallback            -- ^ The producer callback.
              -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.
-producerLoop TopicConnection{..} sensor producer =
+producerLoop topicConnection sensor =
+  let
+    toMessage = makeKeyedMessage (pack sensor) . toStrict . encode
+  in
+    rawProducerLoop topicConnection toMessage
+
+
+-- | Produce messages for a Kafka topic.
+rawProducerLoop :: TopicConnection             -- ^ The Kafka topic name and connection information.
+                -> (a -> Message)              -- ^ The encoder.
+                -> IO [a]                      -- ^ The producer callback.
+                -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.
+rawProducerLoop TopicConnection{..} toMessage producer =
   do
     exitFlag <- newEmptyMVar :: IO (MVar ())
     let
@@ -136,12 +162,12 @@
         do
           events <- liftIO producer
           void
-            $ produceMessages
+            . produceMessages
             $ map
-              (TopicAndMessage (fromString topic) . makeKeyedMessage (pack sensor) . toStrict . encode)
+              (TopicAndMessage (fromString topic) . toMessage)
               events
           running <- liftIO $ isEmptyMVar exitFlag
-          when running
+          when (running && not (null events))
             loop
     return
       (
diff --git a/src/Network/UI/Kafka/Types.hs b/src/Network/UI/Kafka/Types.hs
--- a/src/Network/UI/Kafka/Types.hs
+++ b/src/Network/UI/Kafka/Types.hs
@@ -3,7 +3,7 @@
 Copyright   :  (c) 2016-19 Brian W Bush
 License     :  MIT
 Maintainer  :  Brian W Bush <code@functionally.io>
-Stability   :  Beta
+Stability   :  Production
 Portability :  Portable
 
 Event types.