diff --git a/CHANGES.md b/CHANGES.md
new file mode 100644
--- /dev/null
+++ b/CHANGES.md
@@ -0,0 +1,5 @@
+0.7.0.0
+=======
+
+-   Initial release.
+
diff --git a/Kafka.hs b/Kafka.hs
deleted file mode 100644
--- a/Kafka.hs
+++ /dev/null
@@ -1,163 +0,0 @@
-{-# LANGUAGE NamedFieldPuns #-}
--- |
--- Module      :  Kafka
--- Copyright   :  Abhinav Gupta 2015
--- License     :  MIT
---
--- Maintainer  :  mail@abhinavg.net
--- Stability   :  experimental
--- Portability :  GHC
---
--- A library to interact with Apache Kafka 0.7.
---
-module Kafka
-    (
-    -- * Main interface
-    --
-    -- | Requests to Kafka can be made using 'produce', 'fetch', and
-    -- 'offsets'.  For 'produce' and 'fetch', the functions automatically
-    -- decide whether the request needs to be a single @Produce@/@Fetch@
-    -- request or a @Multi*@ request.
-    --
-    -- The request operations send requests and receive responses using any
-    -- type that is an instance of 'Transport'. 'withConnection' produces one
-    -- such object.
-    --
-      withConnection
-    , produce
-    , fetch
-    , offsets
-
-    -- * Types
-
-    , Produce(..)
-
-    , Fetch(..)
-    , FetchResponse(..)
-
-    , Offsets(..)
-    , OffsetsTime(..)
-
-    , Topic(..)
-    , Offset(..)
-    , Partition(..)
-    , Size(..)
-    , Count(..)
-
-    -- * Other
-
-    , Error(..)
-    , Response
-
-    -- * Transport
-
-    , Socket
-    , Transport(..)
-    ) where
-
-import Control.Applicative
-import Data.Monoid
-
-import qualified Data.Serialize as C
-
-import Kafka.Internal
-
--- | Receives the next response from the connection using the given
--- deserializer.
---
--- Returns either an 'Error' for Kafka errors or the deseriazlied result.
-recvResponse
-    :: Transport t
-    => t        -- ^ Transport from which the response will be read.
-    -> C.Get a  -- ^ Deserializer
-    -> IO (Response a)
-recvResponse transport getBody = do
-    -- TODO this probably belongs in an internal module somewhere
-    -- TODO use protocol error or something for parse errors instead of this
-    -- TODO maybe also catch IO exceptions
-    -- TODO alternatively, throw protocol and IO exceptions instead of failing
-    -- like this.
-    respLen <- C.runGet getLength <$> recv transport 4 >>= either fail return
-    response <- recvExactly transport respLen
-    either fail return $ C.runGet getResponse response
-  where
-    getLength = fromIntegral <$> C.getWord32be
-    getResponse = do
-        err <- C.get
-        body <- getBody
-        return $ maybe (Right body) Left err
-
--- | Sends the given 'Produce' requests to Kafka.
---
--- If multiple requests are supplied, a @MultiProduce@ request is made.
---
--- @
--- 'withConnection' \"localhost\" 9092 $ \\conn ->
---   produce conn [
---      'Produce' ('Topic' \"my-topic\") ('Partition' 0) [\"foo\"]
---    , Produce \"another-topic\" 0
---                [\"multiple\", \"messages\"]
---    ]
--- @
---
--- Note that string literals may be used in place of 'Topic' (with the
--- @OverloadedStrings@ GHC extension), and integer literals may be used in
--- place of 'Partition'.
-produce :: Transport t => t -> [Produce] -> IO ()
-produce transport reqs = case reqs of
-  []  -> return ()
-  [x] -> send transport (C.runPut $ putProduceRequest x)
-  xs  -> send transport (C.runPut $ putMultiProduceRequest xs)
-
--- | 'Fetch'es messages from Kafka.
---
--- If multiple Fetch requests are supplied, a @MultiFetch@ request is made.
---
--- @
--- 'withConnection' \"localhost\" 9092 $ \\conn -> do
---   Right ['FetchResponse' messages newOffset] <- fetch conn [
---       'Fetch' ('Topic' \"test-topic\") ('Partition' 0) (Offset 42) 1024
---     ]
---   {- Consume the messages here -}
---   response <- fetch conn ['Fetch' \"test-topic\" 0 newOffset 1024]
---   {- ... -}
--- @
---
--- Returns a list of 'FetchResponse's in the same order as the 'Fetch'
--- requests. Each response contains the messages returned for the
--- corresponding request and the new offset at which the next request should
--- be made for that request to get the messages that follow.
---
--- If a response for a request contains no messages, the specified
--- topic-partition pair has been exhausted.
---
--- Note that string literals may be used in place of 'Topic' (with the
--- @OverloadedStrings@ GHC extension), and integer literals may be used in
--- place of 'Offset'.
---
-fetch :: Transport t => t -> [Fetch] -> IO (Response [FetchResponse])
-fetch transport reqs = case reqs of
-  [] -> return (Right mempty)
-  [x] -> do
-    send transport (C.runPut $ putFetchRequest x)
-    fmap (:[]) <$> recvResponse transport (getFetchResponse x)
-  xs -> do
-    send transport (C.runPut $ putMultiFetchRequest xs)
-    recvResponse transport (getMultiFetchResponse xs)
-
--- | Retrieve message offsets from Kafka.
---
--- @
--- 'withConnection' \"localhost\" 9092 $ \\conn -> do
---   Right [os] <- offsets conn (Offsets \"topic\" 0 OffsetsEarliest 1)
---   fetch conn ['Fetch' \"topic\" (Partition 0) os 10] >>= doSomething
--- @
---
--- Note that string literals may be used in place of 'Topic' (with the
--- @OverloadedStrings@ GHC extension), and integer literals may be used in
--- place of 'Count'.
---
-offsets :: Transport t => t -> Offsets -> IO (Response [Offset])
-offsets transport req = do
-    send transport (C.runPut $ putOffsetsRequest req)
-    recvResponse transport getOffsetsResponse
diff --git a/Kafka/Internal.hs b/Kafka/Internal.hs
deleted file mode 100644
--- a/Kafka/Internal.hs
+++ /dev/null
@@ -1,11 +0,0 @@
-module Kafka.Internal
-    ( module Kafka.Internal.Request
-    , module Kafka.Internal.Response
-    , module Kafka.Internal.Transport
-    , module Kafka.Internal.Types
-    ) where
-
-import Kafka.Internal.Request
-import Kafka.Internal.Response
-import Kafka.Internal.Transport
-import Kafka.Internal.Types
diff --git a/Kafka/Internal/Request.hs b/Kafka/Internal/Request.hs
deleted file mode 100644
--- a/Kafka/Internal/Request.hs
+++ /dev/null
@@ -1,153 +0,0 @@
-module Kafka.Internal.Request (
-      Produce(..)
-    , putProduceRequest
-    , putMultiProduceRequest
-
-    , Fetch(..)
-    , putFetchRequest
-    , putMultiFetchRequest
-
-    , Offsets(..)
-    , putOffsetsRequest
-    ) where
-
-import Control.Applicative
-import Data.ByteString     (ByteString)
-import Data.Foldable       (Foldable)
-
-import qualified Data.ByteString as B
-import qualified Data.Foldable   as Fold
-import qualified Data.Serialize  as C
-
-import Kafka.Internal.Types
-
--- | 'Put's the items in the given Foldable in-order and prepends the result
--- with a 16-bit integer containing the total number of items that were put.
-putWithCountPrefix :: Foldable f => f a -> C.Putter a -> C.Put
-putWithCountPrefix xs put =
-    let (count, p) = Fold.foldr' go (0, pure ()) xs
-    in C.putWord16be count *> p
-  where
-    go a (c, p) = (c + 1, put a *> p)
-
--- | Prepends the given Put with a 32-bit integer containing the total size of
--- the given Put.
-putWithLengthPrefix :: C.Put -> C.Put
-putWithLengthPrefix p = do
-    C.putWord32be $ fromIntegral (B.length bs)
-    C.putByteString bs
-  where
-    bs = C.runPut p
-
--- | Encodes a request with the given 'RequestType'.
---
--- More specifically, this prepends the given put with the request type and
--- the size of the request.
-encodeRequest :: RequestType -> C.Put -> C.Put
-encodeRequest reqType p = putWithLengthPrefix $ C.put reqType >> p
-
--- | A request to send messages down a Kafka topic-partition pair.
---
--- Produce requests do not have a corresponding response. There is no way of
--- knowing in Kafka 0.7 if a message was successfully @Produce@d.
-data Produce = Produce {
-    produceTopic     :: {-# UNPACK #-} !Topic
-  -- ^ Kafka topic to which the messages will be sent.
-  , producePartition :: {-# UNPACK #-} !Partition
-  -- ^ Partition of the topic.
-  , produceMessages  ::                [ByteString]
-  -- ^ List of message payloads.
-  --
-  -- For those concerned with low-leveld details: These messages will be
-  -- compressed using <https://code.google.com/p/snappy/ Snappy> compression.
-  } deriving (Show, Read, Eq)
-
-encodeProduce :: Produce -> C.Put
-encodeProduce (Produce topic partition messages) = do
-    C.put topic
-    C.put partition
-    putWithLengthPrefix (C.put (MessageSet messages))
-
--- | @Put@s the given single @Produce@ request.
-putProduceRequest :: Produce -> C.Put
-putProduceRequest =
-    encodeRequest ProduceRequestType . encodeProduce
-
--- | @Put@s the given @MultiProduce@ request.
-putMultiProduceRequest :: [Produce] -> C.Put
-putMultiProduceRequest reqs =
-    encodeRequest MultiProduceRequestType $
-        putWithCountPrefix reqs encodeProduce
-
--- | A request to fetch messages from a particular Kafka topic-partition pair.
---
--- 'Kafka.FetchResponse' contains responses for this kind of request.
-data Fetch = Fetch {
-    fetchTopic     :: {-# UNPACK #-} !Topic
-  -- ^ Kafka topic from which messages will be fetched.
-  , fetchPartition :: {-# UNPACK #-} !Partition
-  -- ^ Partition of the topic.
-  , fetchOffset    :: {-# UNPACK #-} !Offset
-  -- ^ Offset at which the fetch will start.
-  --
-  -- Kafka offloads the responsiblity of knowing this to the client. That
-  -- means that if an offset is specified here that is not a real message
-  -- start, Kafka will spit out garbage.
-  --
-  -- Use 'Kafka.offsets' to find valid offsets.
-  , fetchSize      :: {-# UNPACK #-} !Size
-  -- ^ Maximum size of the returned messages.
-  --
-  -- Note, this is /not/ the number of messages. This is the maximum
-  -- combined size of the returned /compressed/ messages.
-  } deriving (Show, Read, Eq)
-
-encodeFetch :: Fetch -> C.Put
-encodeFetch (Fetch topic partition offset maxSize) = do
-    C.put topic
-    C.put partition
-    C.put offset
-    C.put maxSize
-
--- | @Put@s the given single @Fetch@ request.
-putFetchRequest :: Fetch -> C.Put
-putFetchRequest = encodeRequest FetchRequestType . encodeFetch
-
--- | @Put@s the given @MultiFetch@ request.
-putMultiFetchRequest :: [Fetch] -> C.Put
-putMultiFetchRequest reqs =
-    encodeRequest MultiFetchRequestType $
-        putWithCountPrefix reqs encodeFetch
-
--- | A request to retrieve offset information from Kafka.
---
--- The response for this kind of request is a list of 'Offset's.
-data Offsets = Offsets {
-    offsetsTopic     :: {-# UNPACK #-} !Topic
-  -- ^ Kafka topic from which offsets will be retrieved.
-  , offsetsPartition :: {-# UNPACK #-} !Partition
-  -- ^ Partition of the topic.
-  , offsetsTime      ::                !OffsetsTime
-  -- ^ Time around which offsets will be retrieved.
-  --
-  -- If you provide a time for this, keep in mind that the response will not
-  -- contain the precise offset that occurred around that time. It will return
-  -- up to @offsetsCount@ offsets in descending, each being the first offset
-  -- of every segment file for the specified partition with a modified time
-  -- less than the specified time, and possibly a "high water mark" for the
-  -- last segment of the partition (if it was modified before the specified
-  -- time) which specifies the offset at which the next message to that
-  -- partition will be written.
-  , offsetsCount     :: {-# UNPACK #-} !Count
-  -- ^ Maximum number of offsets that will be retrieved.
-  } deriving (Show, Read, Eq)
-
--- | @Put@s the given @Offsets@ request.
-putOffsetsRequest :: Offsets -> C.Put
-putOffsetsRequest (Offsets topic partition time maxNumber) =
-    encodeRequest OffsetsRequestType $ do
-        C.put topic
-        C.put partition
-        C.put time
-        C.put maxNumber
-
diff --git a/Kafka/Internal/Response.hs b/Kafka/Internal/Response.hs
deleted file mode 100644
--- a/Kafka/Internal/Response.hs
+++ /dev/null
@@ -1,61 +0,0 @@
-module Kafka.Internal.Response
-    ( Response
-    , FetchResponse(..)
-    , getFetchResponse
-    , getMultiFetchResponse
-    , getOffsetsResponse
-    ) where
-
-import Control.Applicative
-import Control.Monad
-import Data.ByteString     (ByteString)
-import Data.Monoid
-
-import qualified Data.Serialize as C
-
-import Kafka.Internal.Request
-import Kafka.Internal.Types
-
--- | A response from Kafka can either be a failure or the value that was
--- expected.
-type Response a = Either Error a
-
--- | Result of a single Kafka 'Kafka.Fetch'.
-data FetchResponse = FetchResponse {
-    fetchMessages  ::                [ByteString]
-  -- ^ List of messages returned in the response for the 'Kafka.Fetch'
-  -- request.
-  , fetchNewOffset :: {-# UNPACK #-} !Offset
-  -- ^ New offset at which the next 'Fetch' request should start reading in
-  -- the same topic and partition to access the messages that follow the
-  -- messages returned in this response.
-  } deriving (Show, Read, Eq)
-
--- | Parses a single @FetchResponse@ for the given @Fetch@ request.
-getFetchResponse :: Fetch -> C.Get FetchResponse
-getFetchResponse Fetch{fetchOffset = oldOffset} = do
-    startBytes <- C.remaining
-    messages <- fromMessageSet . mconcat <$> many C.get
-    endBytes <- C.remaining
-    let newOffset = oldOffset + fromIntegral (startBytes - endBytes)
-    return $ FetchResponse messages $! newOffset
-
--- | Parses @FetchResponse@s for each of the given @Fetch@ requests.
-getMultiFetchResponse :: [Fetch] -> C.Get [FetchResponse]
-getMultiFetchResponse requests = forM requests $ \request -> do
-    size <- fromIntegral <$> C.getWord32be
-    C.skip 2 -- skip error
-    isolate' (size - 2) (getFetchResponse request)
-
--- | Parses the response for an @Offsets@ request.
-getOffsetsResponse :: C.Get [Offset]
-getOffsetsResponse = do
-    numOffsets <- C.getWord32be
-    replicateM (fromIntegral numOffsets) C.get
-
--- | A version of 'C.isolate' that does not fail the parse because of
--- unconsumed data.
-isolate' :: Int -> C.Get a -> C.Get a
-isolate' n m = do
-    s <- C.getBytes n
-    either fail return $ C.runGet m s
diff --git a/Kafka/Internal/Transport.hs b/Kafka/Internal/Transport.hs
deleted file mode 100644
--- a/Kafka/Internal/Transport.hs
+++ /dev/null
@@ -1,88 +0,0 @@
-{-# LANGUAGE NamedFieldPuns #-}
-module Kafka.Internal.Transport
-    ( Transport(..)
-    , recvExactly
-
-    , Socket
-    , withConnection
-    , connect
-    , close
-    ) where
-
-
-import Control.Applicative
-import Data.ByteString     (ByteString)
-import Network.Socket      (Socket, close)
-
-import qualified Control.Exception         as E
-import qualified Data.ByteString           as B
-import qualified Data.ByteString.Char8     as B8
-import qualified Network.Socket            as N
-import qualified Network.Socket.ByteString as S
-
--- | Types that provide a means to send and receive bytes.
-class Transport t where
-    -- | Send the given ByteString down the transport.
-    --
-    -- This must block until the request has been finished.
-    send :: t -> ByteString -> IO ()
-    -- | Read up to the given number of bytes from the stream.
-    --
-    -- The returned ByteString may be empty if the end of the stream was
-    -- reached.
-    recv :: t -> Int -> IO ByteString
-
-instance Transport Socket where
-    send = S.sendAll
-    recv = S.recv
-
--- | Keep reading from the given transport until the given number of bytes are
--- read or the end of the stream is reached -- whichever comes first.
-recvExactly :: Transport t => t -> Int -> IO ByteString
-recvExactly t size = B.concat . reverse <$> loop [] 0
-  where
-    loop chunks bytesRead
-        | bytesRead >= size = return chunks
-        | otherwise = do
-            chunk <- recv t (size - bytesRead)
-            if B.null chunk
-              then return chunks
-              else loop (chunk:chunks) $! bytesRead + B.length chunk
- -- TODO this should return a bytestring of exactly the given size. Anything
- -- else should be an IO exception or a protocol error since we use this only
- -- in cases where we know the exact size of the response.
-
-getAddrInfo :: ByteString -> Int -> IO N.AddrInfo
-getAddrInfo host port = head <$>
-    N.getAddrInfo (Just hints)
-                  (Just $ B8.unpack host)
-                  (Just $ show port)
-  where
-    hints = N.defaultHints {
-        N.addrFamily = N.AF_INET
-      , N.addrFlags = [N.AI_NUMERICSERV]
-      , N.addrSocketType = N.Stream
-      }
-
--- | Create a new connection.
---
--- Connects to the given hostname and port. Throws an 'N.IOException' in case
--- of failure.
-connect :: ByteString -> Int -> IO Socket
-connect host port = do
-    addrInfo <- getAddrInfo host port
-    socket <- N.socket N.AF_INET N.Stream N.defaultProtocol
-    N.connect socket (N.addrAddress addrInfo) `E.onException` N.close socket
-    return socket
-
--- | Open a connection, execute the given operation on it, and ensure it is
--- closed afterwards even if an exception was thrown.
---
--- > withConnection "localhost" 9092 $ \conn ->
--- >    doStuff conn
--- >    fail "something went wrong"
---
--- Throws an 'IOException' if we were unable to open the connection.
---
-withConnection :: ByteString -> Int -> (Socket -> IO a) -> IO a
-withConnection host port = E.bracket (connect host port) close
diff --git a/Kafka/Internal/Types.hs b/Kafka/Internal/Types.hs
deleted file mode 100644
--- a/Kafka/Internal/Types.hs
+++ /dev/null
@@ -1,289 +0,0 @@
-{-# LANGUAGE FlexibleInstances          #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE OverlappingInstances       #-}
-module Kafka.Internal.Types
-    ( Error(..)
-    , Compression(..)
-    , OffsetsTime(..)
-    , RequestType(..)
-
-    , Topic(..)
-    , Offset(..)
-    , Partition(..)
-    , Size(..)
-    , Count(..)
-
-    , Message(..)
-    , MessageSet(..)
-    ) where
-
-import Control.Applicative
-import Data.ByteString       (ByteString)
-import Data.ByteString.Lazy  (fromStrict, toStrict)
-import Data.Digest.CRC32     (crc32)
-import Data.DList            (DList)
-import Data.Monoid
-import Data.String
-import Data.Time             (UTCTime)
-import Data.Time.Clock.POSIX
-import Data.Word
-
-import qualified Codec.Compression.GZip   as GZip
-import qualified Codec.Compression.Snappy as Snappy
-import qualified Data.ByteString          as B
-import qualified Data.DList               as DList
-import qualified Data.Serialize           as C
-
--- | Different errors returned by Kafka.
-data Error
-    = UnknownError          -- -1
-    -- ^ Unknown error
-    | OffsetOutOfRangeError --  1
-    -- ^ Offset requested is invalid or no longer available on the server.
-    | InvalidMessageError   --  2
-    -- ^ A message failed to match its checksum.
-    | WrongPartitionError   --  3
-    -- ^ The requested partition doesn't exist.
-    | InvalidFetchSizeError --  4
-    -- ^ The maximum size requested for fetching is smaller than the message
-    -- being fetched.
-  deriving (Show, Read, Eq, Ord)
-
-instance C.Serialize (Maybe Error) where
-    -- A note about extensions:
-    -- We need FlexibleInstances to allow GHC to let us declare an instance
-    -- for the type @Maybe Error@. We also need OverlappingInstances to tell
-    -- it to pick this (more specific) instance of Maybe Error over the
-    -- generic one provided by cereal.
-    get = do
-        code <- C.getWord16be
-        case code of
-            0  -> return Nothing
-            1  -> return $ Just OffsetOutOfRangeError
-            2  -> return $ Just InvalidMessageError
-            3  -> return $ Just WrongPartitionError
-            4  -> return $ Just InvalidFetchSizeError
-            -1 -> return $ Just UnknownError
-            _  ->   fail $ "Unknown error code: " ++ show code
-
-    put                     Nothing  = C.putWord16be 0
-    put (Just OffsetOutOfRangeError) = C.putWord16be 1
-    put (Just   InvalidMessageError) = C.putWord16be 2
-    put (Just   WrongPartitionError) = C.putWord16be 3
-    put (Just InvalidFetchSizeError) = C.putWord16be 4
-    put (Just          UnknownError) = C.putWord16be (-1)
-
--- | Methods of compression supported by Kafka.
-data Compression
-    = NoCompression
-    -- ^ The message is uncompressed.
-    | GzipCompression
-    -- ^ The message is compressed using @gzip@ compression and may contain
-    -- other messages in it.
-    | SnappyCompression
-    -- ^ The message is compressed using @snappy@ compression and may contain
-    -- other messages in it.
-  deriving (Show, Read, Eq, Ord)
-
-instance C.Serialize Compression where
-    put NoCompression     = C.putWord8 0
-    put GzipCompression   = C.putWord8 1
-    put SnappyCompression = C.putWord8 2
-
-    get = C.getWord8 >>= \i -> case i of
-        0 -> return NoCompression
-        1 -> return GzipCompression
-        2 -> return SnappyCompression
-        _ -> fail $ "Invalid compression code: " ++ show i
-
--- | Different times for which offsets may be retrieved using
--- 'Kafka.offsets'.
-data OffsetsTime
-    = OffsetsLatest
-    -- ^ Retrieve the latest offsets
-    | OffsetsEarliest
-    -- ^ Retrieve the earliest offsets.
-    | OffsetsBefore !UTCTime
-    -- ^ Retrieve offsets before the given time.
-    --
-    -- Keep in mind that the response will not contain the precise offset that
-    -- occurred around this time. It will return up to the specified count of
-    -- offsets in descending, each being the first offset of every segment
-    -- file for the specified partition with a modified time less than this
-    -- time, and possibly a "high water mark" for the last segment of the
-    -- partition (if it was modified before this time) which specifies the
-    -- offset at which the next message to that partition will be written.
-  deriving (Show, Read, Eq, Ord)
-
-instance C.Serialize OffsetsTime where
-    put OffsetsLatest = C.putWord64be (-1)
-    put OffsetsEarliest = C.putWord64be (-2)
-    put (OffsetsBefore t) =
-        C.putWord64be . round . (* 1000) . utcTimeToPOSIXSeconds $ t
-
-    -- Kafka doesn't care about anything beyond milliseconds while UTCTime can
-    -- be more precise. We will lose some amount of precision if the same
-    -- UTCTime is serialized and then deserialized.
-    get = C.getWord64be >>= \i -> case i of
-        -1 -> return   OffsetsLatest
-        -2 -> return   OffsetsEarliest
-        t  -> return . OffsetsBefore .
-               posixSecondsToUTCTime . (/ 1000) . realToFrac $ t
-
--- | The different request types supported by Kafka.
-data RequestType
-    = ProduceRequestType        -- 0
-    | FetchRequestType          -- 1
-    | MultiFetchRequestType     -- 2
-    | MultiProduceRequestType   -- 3
-    | OffsetsRequestType        -- 4
-  deriving (Show, Read, Eq, Ord)
-
-instance C.Serialize RequestType where
-    put ProduceRequestType      = C.putWord16be 0
-    put FetchRequestType        = C.putWord16be 1
-    put MultiFetchRequestType   = C.putWord16be 2
-    put MultiProduceRequestType = C.putWord16be 3
-    put OffsetsRequestType      = C.putWord16be 4
-
-    get = C.getWord16be >>= \i -> case i of
-            0 -> return ProduceRequestType
-            1 -> return FetchRequestType
-            2 -> return MultiFetchRequestType
-            3 -> return MultiProduceRequestType
-            4 -> return OffsetsRequestType
-            _ -> fail $ "Unknown request type: " ++ show i
-
--- | Represents a Kafka topic.
---
--- This is an instance of 'IsString' so a literal string may be used to create
--- a Topic with the @OverloadedStrings@ extension.
-newtype Topic = Topic ByteString
-    deriving (Show, Read, Eq, Ord, IsString)
-
-instance C.Serialize Topic where
-    put (Topic topic) = do
-        C.putWord16be (fromIntegral $ B.length topic)
-        C.putByteString topic
-    get = do
-        topicLength <- fromIntegral <$> C.getWord16be
-        Topic <$> C.getByteString topicLength
-
--- | Represents an Offset in Kafka.
---
--- This is an instance of 'Num' so a literal number may be used to create an
--- Offset.
-newtype Offset = Offset Word64
-    deriving (Show, Read, Eq, Ord, Num)
-
-instance C.Serialize Offset where
-    put (Offset o) = C.putWord64be o
-    get = Offset <$> C.getWord64be
-
--- | Represents a Kafka topic partition.
---
--- This is an instance of 'Num' so a literal number may be used to create a
--- Partition.
-newtype Partition = Partition Word32
-    deriving (Show, Read, Eq, Ord, Num)
-
-instance C.Serialize Partition where
-    put (Partition p) = C.putWord32be p
-    get = Partition <$> C.getWord32be
-
--- | Represents a size.
---
--- This is an instance of 'Num' so a literal number may be used to create a
--- Size.
-newtype Size = Size Word32
-    deriving (Show, Read, Eq, Ord, Num)
-
-instance C.Serialize Size where
-    put (Size s) = C.putWord32be s
-    get = Size <$> C.getWord32be
-
--- | Represents a Count.
---
--- This is an instance of 'Num' so a literal number may be used to create a
--- Size.
-newtype Count = Count Word32
-    deriving (Show, Read, Eq, Ord, Num)
-
-instance C.Serialize Count where
-    put (Count s) = C.putWord32be s
-    get = Count <$> C.getWord32be
-
--- | Represents a Message being sent through Kafka.
-data Message = Message {
-    messageCompression :: !Compression
-  -- ^ Compression used for the message.
-  --
-  -- If this is anything but 'NoCompression', this message contains other
-  -- messages in it.
-  , messagePayload     :: !ByteString
-  -- ^ Message payload.
-  --
-  -- If the message is using any compression scheme, the payload contains
-  -- other messages in the same format.
-  } deriving (Show, Read, Eq, Ord)
-
-instance C.Serialize Message where
-    put (Message compression payload) = do
-        C.putWord32be (fromIntegral messageLength)
-        C.putWord8 1 -- Magic
-        C.put compression
-        C.putWord32be (crc32 payload)
-        C.putByteString payload
-      where
-        messageLength = B.length payload
-            + 4 -- checksum
-            + 1 -- compression
-            + 1 -- magic
-
-    get = C.getWord32be >>= \messageLength -> do
-        magic <- C.getWord8
-        compression <- case magic of
-            0 -> return NoCompression
-            1 -> C.get
-            _ -> fail $ "Unknown magic code: " ++ show magic
-        checksum <- C.getWord32be
-        let remainingLength = fromIntegral $
-              messageLength -
-                ( 4                  -- checksum
-                + 1                  -- magic
-                + fromIntegral magic -- compression (1 byte if magic is 1)
-                )
-        payload <- C.getByteString remainingLength
-        if crc32 payload == checksum
-          then return (Message compression payload)
-          else fail "Checksum did not match."
-            -- TODO: This should probably be InvalidMessageError
-
--- | Represents a collection of message payloads.
---
--- These are compressed into a single message using Snappy compression when
--- being sent.
-newtype MessageSet = MessageSet { fromMessageSet :: [ByteString] }
-    deriving (Show, Read, Eq, Monoid)
-
-instance C.Serialize MessageSet where
-    put (MessageSet messages) = C.put (Message SnappyCompression payload)
-      where
-        payload = Snappy.compress . C.runPut $
-                    mapM_ (C.put . Message NoCompression) messages
-
-    get = MessageSet . DList.toList <$> (C.get >>= readMessages)
-      where
-        readMessages :: Message -> C.Get (DList ByteString)
-        readMessages (Message NoCompression payload) =
-            return (DList.singleton payload)
-        readMessages (Message compression payload) = do
-            messages <- either fail return $
-                        C.runGet (many C.get) decompressedPayload
-            mconcat <$> mapM readMessages messages
-          where
-            decompressedPayload = decompress payload
-            decompress = case compression of
-              NoCompression -> id
-              SnappyCompression -> Snappy.decompress
-              GzipCompression -> toStrict . GZip.decompress . fromStrict
diff --git a/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -0,0 +1,27 @@
+`kafka-client`
+==============
+
+This package provides a low-level Haskell client for [Apache Kafka][] 0.7.
+**Kafka 0.8 and newer are not yet supported.**
+
+Package documentation is available on [Hackage][] and [here][].
+
+<dl>
+<dt>
+Why Kafka 0.7?
+</dt>
+<dd>
+I needed to interact with a system running Kafka 0.7.
+</dd>
+<dt>
+Will 0.8 or newer be supported?
+</dt>
+<dd>
+There are currently no plans to do so. Unless I end up needing to also interact
+with a Kafka 0.8 system, or receive a pull request, the answer is no.
+</dd>
+</dl>
+
+  [Apache Kafka]: http://kafka.apache.org/
+  [Hackage]: http://hackage.haskell.org/package/kafka-client
+  [here]: http://abhinavg.net/kafka-client/
diff --git a/examples/consoleConsumer.hs b/examples/consoleConsumer.hs
new file mode 100644
--- /dev/null
+++ b/examples/consoleConsumer.hs
@@ -0,0 +1,36 @@
+{-# LANGUAGE OverloadedStrings #-}
+module Main (main) where
+
+import           Control.Concurrent    (threadDelay)
+import           Control.Monad
+import qualified Data.ByteString.Char8 as B8
+import           System.Environment
+import           System.Exit
+
+import Kafka
+
+main :: IO ()
+main = do
+  args <- getArgs
+  when (null args) $ do
+    putStrLn "USAGE: consoleProducer TOPIC"
+    exitFailure
+
+  let topic = Topic (B8.pack $ head args)
+
+  withConnection "localhost" 9092 $ \conn -> do
+    startOffset:_ <- offsets conn (Offsets topic 0 OffsetsLatest 10)
+                        >>= either (fail . show) return
+    loop conn (Fetch topic 0 startOffset (1024 * 1024))
+  where
+    loop conn fetchReq = do
+        [FetchResponse{
+            fetchMessages = messages
+          , fetchNewOffset = newOffset
+          }] <- fetch conn [fetchReq] >>= either (fail . show) return
+
+        mapM_ B8.putStrLn messages
+        when (null messages) $
+            threadDelay (1000 * 1000)
+
+        loop conn fetchReq{fetchOffset = newOffset}
diff --git a/examples/consoleProducer.hs b/examples/consoleProducer.hs
new file mode 100644
--- /dev/null
+++ b/examples/consoleProducer.hs
@@ -0,0 +1,31 @@
+{-# LANGUAGE OverloadedStrings #-}
+module Main (main) where
+
+import           Control.Applicative
+import           Control.Monad
+import qualified Data.ByteString.Char8 as B8
+import           System.Environment
+import           System.Exit
+import qualified System.IO             as IO
+
+import Kafka
+
+main :: IO ()
+main = do
+  args <- getArgs
+  when (null args) $ do
+    putStrLn "USAGE: consoleProducer TOPIC"
+    exitFailure
+
+  let topic = Topic (B8.pack $ head args)
+  withConnection "localhost" 9092 $ \conn ->
+    whileM_ (not <$> IO.isEOF) $ do
+        line <- B8.getLine
+        produce conn [Produce topic 0 [line]]
+
+whileM_ :: Monad m => m Bool -> m a -> m ()
+whileM_ cond action = loop
+  where
+    loop = do
+        x <- cond
+        when x (action >> loop)
diff --git a/integration-test/IntegrationTest.hs b/integration-test/IntegrationTest.hs
new file mode 100644
--- /dev/null
+++ b/integration-test/IntegrationTest.hs
@@ -0,0 +1,30 @@
+{-# LANGUAGE OverloadedStrings #-}
+module Main (main) where
+
+import Control.Exception
+import Test.Hspec
+import Test.Kafka.Managed
+
+import qualified Network        as N
+import qualified System.IO.Temp as Temp
+
+import qualified Kafka
+import qualified KafkaSpec
+
+getRandomUnusedPort :: IO Int
+getRandomUnusedPort =
+    bracket (N.listenOn $ N.PortNumber 0) N.sClose $ \socket -> do
+        (N.PortNumber num) <- N.socketPort socket
+        return (fromIntegral num)
+
+main :: IO ()
+main = do
+  kafkaPort <- getRandomUnusedPort
+  Temp.withSystemTempDirectory "kafka-logs" $ \logDir ->
+    let kafkaConfig = KafkaConfig {
+            kafkaLogDirectory = logDir
+          , kafkaServerPort = kafkaPort
+          }
+    in withManagedKafka kafkaConfig $
+       Kafka.withConnection "localhost" kafkaPort $
+           hspec . KafkaSpec.spec
diff --git a/integration-test/KafkaSpec.hs b/integration-test/KafkaSpec.hs
new file mode 100644
--- /dev/null
+++ b/integration-test/KafkaSpec.hs
@@ -0,0 +1,134 @@
+{-# LANGUAGE OverloadedStrings #-}
+module KafkaSpec (spec) where
+
+import Control.Applicative   ((<$>))
+import Control.Concurrent    (threadDelay)
+import Control.Monad         (replicateM)
+import Data.IORef            (modifyIORef', newIORef, readIORef)
+import Test.Hspec
+import Test.Hspec.QuickCheck (prop)
+import Test.QuickCheck
+
+import qualified Data.Set as Set
+
+import Kafka
+import Test.Kafka.Arbitrary
+
+-- | Keep running the given generator until a value that satisfies the given
+-- predicate is produced.
+generateSuchThat :: Gen a -> (a -> IO Bool) -> IO a
+generateSuchThat gen f = loop
+  where
+    loop = do
+      a <- generate gen
+      match <- f a
+      if match
+        then return a
+        else loop
+
+-- | Returns an IO action that will always return unique values from the given
+-- generator.
+uniqueGenerator :: Ord a => Gen a -> IO (IO a)
+uniqueGenerator gen = do
+    generated <- newIORef Set.empty
+    let isNew a = Set.notMember a <$> readIORef generated
+    return $ do
+        a <- generateSuchThat gen isNew
+        modifyIORef' generated (Set.insert a)
+        return a
+
+spec :: Socket -> Spec
+spec conn = describe "Kafka" $ do
+
+    -- This is hacky but it will ensure that none of the tests run under the
+    -- same Kafka topic.
+    getNewTopic <- runIO $ uniqueGenerator arbitrary
+
+    prop "returns zero offset for empty topic" $ do
+        topic <- getNewTopic
+        let req = Offsets topic 0 OffsetsLatest 10
+        offsets conn req `shouldReturn` Right [0]
+
+    prop "can produce and fetch empty payloads" $ do
+        topic <- getNewTopic
+        produce conn [Produce topic 0 [""]]
+        Right resp <- pollMessages conn $ Fetch topic 0 0 (1024 * 1024)
+        fetchMessages resp `shouldBe` [""]
+
+    prop "can produce and fetch empty lists of payloads" $ do
+        topic <- getNewTopic
+        produce conn [Produce topic 0 []]
+        fetch conn [fetchRequest topic 0]
+            `shouldReturn` Right [FetchResponse [] 0]
+
+    prop "can fetch and produce" $ \(NonEmptyPayloadList messages) -> do
+        topic <- getNewTopic
+        produce conn [Produce topic 0 messages]
+
+        Right resp <- pollMessages conn (fetchRequest topic 0)
+        fetchMessages resp `shouldBe` messages
+
+        let newOffset = fetchNewOffset resp
+
+        offsets conn (Offsets topic 0 OffsetsLatest 10)
+            `shouldReturn` Right [newOffset, 0]
+
+        fetch conn [fetchRequest topic newOffset]
+            `shouldReturn` Right [FetchResponse [] newOffset]
+
+    prop "can multi-fetch and multi-produce" $ \(NonEmpty ps) -> do
+        topics <- replicateM (length ps) getNewTopic
+        let messageSets = map getPayloads ps
+            produceReqs = zipWith
+                            (\t ms -> Produce t 0 ms) topics messageSets
+            fetchReqs = map (`fetchRequest` 0) topics
+
+            receivedAll (Left _) = False
+            receivedAll (Right responses) = all isSuccess responses
+              where
+                isSuccess = not . null . fetchMessages
+
+        produce conn produceReqs
+        responses <- do
+            result <- retryUntil receivedAll 10 500 (fetch conn fetchReqs)
+            case result of
+                Just (Right responses) -> return responses
+                Just (Left e) -> fail (show e)
+                Nothing -> fail "Failed to fetch messages."
+
+        map fetchMessages responses `shouldBe` messageSets
+  where
+    fetchRequest topic offset = Fetch topic 0 offset (1024 * 1024)
+
+-- | Polls the given transport with the given Fetch request until a non-empty
+-- list of messages is available.
+--
+-- Throws an error if a non-empty list of messages is not received after
+-- trying for 5 seconds.
+pollMessages :: Transport t => t -> Fetch -> IO (Response FetchResponse)
+pollMessages t req = do
+    result <- retryUntil isSuccess 10 500 (fetch t [req])
+    case result of
+        Nothing -> fail "Failed to fetch messages."
+        Just xs -> return $ head <$> xs
+  where
+    isSuccess (Left _) = False
+    isSuccess (Right [resp]) = not . null . fetchMessages $ resp
+    isSuccess (Right xs) =
+        error $ "Too many messages in response: " ++ show xs
+
+-- | Retries the given IO operation with the specified delay until the given
+-- predicate on the result returns True.
+--
+-- Return Nothing if the maximum number of attempts was exceeded.
+retryUntil :: (a -> Bool) -> Int -> Int -> IO a -> IO (Maybe a)
+retryUntil predicate maxAttempts delayMillis m = loop 0
+  where
+    loop attempt
+        | attempt >= maxAttempts = return Nothing
+        | otherwise = do
+            a <- m
+            if predicate a
+              then return (Just a)
+              else threadDelay (delayMillis * 1000) >> loop (attempt + 1)
+
diff --git a/kafka-client.cabal b/kafka-client.cabal
--- a/kafka-client.cabal
+++ b/kafka-client.cabal
@@ -1,7 +1,7 @@
 name          : kafka-client
-version       : 0.7.0.0
+version       : 0.7.0.1
 synopsis      : Low-level Haskell client library for Apache Kafka 0.7.
-homepage      : https://github.com/abhinav/haskell-kafka-client
+homepage      : https://github.com/abhinav/kafka-client
 license       : MIT
 license-file  : LICENSE
 author        : Abhinav Gupta
@@ -14,6 +14,10 @@
     with <http://kafka.apache.org/ Apache Kafka> 0.7.
     .
     __Kafka 0.8 and newer are not yet supported.__
+extra-source-files:
+    README.md
+    CHANGES.md
+    examples/*.hs
 
 library
   exposed-modules  : Kafka
@@ -22,8 +26,10 @@
                    , Kafka.Internal.Response
                    , Kafka.Internal.Transport
                    , Kafka.Internal.Types
+  hs-source-dirs   : src
   ghc-options      : -Wall
-  build-depends    : base       >= 4.7   && < 4.8
+  build-depends    : base       >= 4.7   && < 4.9
+
                    , bytestring >= 0.10
                    , cereal     >= 0.4
                    , digest     >= 0.0.1
@@ -35,17 +41,23 @@
 
                    -- Test suite-only dependencies:
                    -- Workaround for ghc-mod #339. Uncomment while developing.
-                   -- , hspec              >= 2.0
-                   -- , hspec-discover     >= 2.1
-                   -- , QuickCheck         >= 2.5
+                   -- , containers
+                   -- , hspec
+                   -- , hspec-discover
+                   -- , process
+                   -- , temporary
+                   -- , QuickCheck
   default-language : Haskell2010
 
 test-suite spec
   type             : exitcode-stdio-1.0
   ghc-options      : -Wall
-  hs-source-dirs   : test
+  hs-source-dirs   : test test-lib
   main-is          : Spec.hs
+  other-modules    : Kafka.Internal.TypesSpec
+                   , Test.Kafka.Arbitrary
   build-depends    : base
+
                    , bytestring
                    , cereal
                    , hspec          >= 2.0
@@ -56,6 +68,39 @@
                    , kafka-client
   default-language : Haskell2010
 
+flag integration
+  default     : False
+  description : Enable integration tests.
+
+test-suite integration-test
+  if flag(integration)
+    buildable      : True
+  else
+    buildable      : False
+  type             : exitcode-stdio-1.0
+  ghc-options      : -Wall
+  hs-source-dirs   : integration-test test-lib
+  main-is          : IntegrationTest.hs
+  other-modules    : KafkaSpec
+                   , Test.Kafka.Arbitrary
+                   , Test.Kafka.Managed
+  build-depends    : base
+
+                   , bytestring
+                   , cereal
+                   , containers
+                   , hspec          >= 2.0
+                   , hspec-discover >= 2.1
+                   , network        >= 2.4
+                   , process        >= 1.2
+                   , QuickCheck     >= 2.5
+                   , temporary      >= 1.2
+                   , time
+
+                   , kafka-client
+  default-language : Haskell2010
+
+
 source-repository head
     type: git
-    location: git://github.com/abhinav/haskell-kafka-client.git
+    location: git://github.com/abhinav/kafka-client.git
diff --git a/src/Kafka.hs b/src/Kafka.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka.hs
@@ -0,0 +1,163 @@
+{-# LANGUAGE NamedFieldPuns #-}
+-- |
+-- Module      :  Kafka
+-- Copyright   :  Abhinav Gupta 2015
+-- License     :  MIT
+--
+-- Maintainer  :  mail@abhinavg.net
+-- Stability   :  experimental
+-- Portability :  GHC
+--
+-- A library to interact with Apache Kafka 0.7.
+--
+module Kafka
+    (
+    -- * Main interface
+    --
+    -- | Requests to Kafka can be made using 'produce', 'fetch', and
+    -- 'offsets'.  For 'produce' and 'fetch', the functions automatically
+    -- decide whether the request needs to be a single @Produce@/@Fetch@
+    -- request or a @Multi*@ request.
+    --
+    -- The request operations send requests and receive responses using any
+    -- type that is an instance of 'Transport'. 'withConnection' produces one
+    -- such object.
+    --
+      withConnection
+    , produce
+    , fetch
+    , offsets
+
+    -- * Types
+
+    , Produce(..)
+
+    , Fetch(..)
+    , FetchResponse(..)
+
+    , Offsets(..)
+    , OffsetsTime(..)
+
+    , Topic(..)
+    , Offset(..)
+    , Partition(..)
+    , Size(..)
+    , Count(..)
+
+    -- * Other
+
+    , Error(..)
+    , Response
+
+    -- * Transport
+
+    , Socket
+    , Transport(..)
+    ) where
+
+import Control.Applicative
+import Data.Monoid
+
+import qualified Data.Serialize as C
+
+import Kafka.Internal
+
+-- | Receives the next response from the connection using the given
+-- deserializer.
+--
+-- Returns either an 'Error' for Kafka errors or the deseriazlied result.
+recvResponse
+    :: Transport t
+    => t        -- ^ Transport from which the response will be read.
+    -> C.Get a  -- ^ Deserializer
+    -> IO (Response a)
+recvResponse transport getBody = do
+    -- TODO this probably belongs in an internal module somewhere
+    -- TODO use protocol error or something for parse errors instead of this
+    -- TODO maybe also catch IO exceptions
+    -- TODO alternatively, throw protocol and IO exceptions instead of failing
+    -- like this.
+    respLen <- C.runGet getLength <$> recv transport 4 >>= either fail return
+    response <- recvExactly transport respLen
+    either fail return $ C.runGet getResponse response
+  where
+    getLength = fromIntegral <$> C.getWord32be
+    getResponse = do
+        err <- C.get
+        body <- getBody
+        return $ maybe (Right body) Left err
+
+-- | Sends the given 'Produce' requests to Kafka.
+--
+-- If multiple requests are supplied, a @MultiProduce@ request is made.
+--
+-- @
+-- 'withConnection' \"localhost\" 9092 $ \\conn ->
+--   produce conn [
+--      'Produce' ('Topic' \"my-topic\") ('Partition' 0) [\"foo\"]
+--    , Produce \"another-topic\" 0
+--                [\"multiple\", \"messages\"]
+--    ]
+-- @
+--
+-- Note that string literals may be used in place of 'Topic' (with the
+-- @OverloadedStrings@ GHC extension), and integer literals may be used in
+-- place of 'Partition'.
+produce :: Transport t => t -> [Produce] -> IO ()
+produce transport reqs = case reqs of
+  []  -> return ()
+  [x] -> send transport (C.runPut $ putProduceRequest x)
+  xs  -> send transport (C.runPut $ putMultiProduceRequest xs)
+
+-- | 'Fetch'es messages from Kafka.
+--
+-- If multiple Fetch requests are supplied, a @MultiFetch@ request is made.
+--
+-- @
+-- 'withConnection' \"localhost\" 9092 $ \\conn -> do
+--   Right ['FetchResponse' messages newOffset] <- fetch conn [
+--       'Fetch' ('Topic' \"test-topic\") ('Partition' 0) (Offset 42) 1024
+--     ]
+--   {- Consume the messages here -}
+--   response <- fetch conn ['Fetch' \"test-topic\" 0 newOffset 1024]
+--   {- ... -}
+-- @
+--
+-- Returns a list of 'FetchResponse's in the same order as the 'Fetch'
+-- requests. Each response contains the messages returned for the
+-- corresponding request and the new offset at which the next request should
+-- be made for that request to get the messages that follow.
+--
+-- If a response for a request contains no messages, the specified
+-- topic-partition pair has been exhausted.
+--
+-- Note that string literals may be used in place of 'Topic' (with the
+-- @OverloadedStrings@ GHC extension), and integer literals may be used in
+-- place of 'Offset'.
+--
+fetch :: Transport t => t -> [Fetch] -> IO (Response [FetchResponse])
+fetch transport reqs = case reqs of
+  [] -> return (Right mempty)
+  [x] -> do
+    send transport (C.runPut $ putFetchRequest x)
+    fmap (:[]) <$> recvResponse transport (getFetchResponse x)
+  xs -> do
+    send transport (C.runPut $ putMultiFetchRequest xs)
+    recvResponse transport (getMultiFetchResponse xs)
+
+-- | Retrieve message offsets from Kafka.
+--
+-- @
+-- 'withConnection' \"localhost\" 9092 $ \\conn -> do
+--   Right [os] <- offsets conn (Offsets \"topic\" 0 OffsetsEarliest 1)
+--   fetch conn ['Fetch' \"topic\" (Partition 0) os 10] >>= doSomething
+-- @
+--
+-- Note that string literals may be used in place of 'Topic' (with the
+-- @OverloadedStrings@ GHC extension), and integer literals may be used in
+-- place of 'Count'.
+--
+offsets :: Transport t => t -> Offsets -> IO (Response [Offset])
+offsets transport req = do
+    send transport (C.runPut $ putOffsetsRequest req)
+    recvResponse transport getOffsetsResponse
diff --git a/src/Kafka/Internal.hs b/src/Kafka/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Internal.hs
@@ -0,0 +1,11 @@
+module Kafka.Internal
+    ( module Kafka.Internal.Request
+    , module Kafka.Internal.Response
+    , module Kafka.Internal.Transport
+    , module Kafka.Internal.Types
+    ) where
+
+import Kafka.Internal.Request
+import Kafka.Internal.Response
+import Kafka.Internal.Transport
+import Kafka.Internal.Types
diff --git a/src/Kafka/Internal/Request.hs b/src/Kafka/Internal/Request.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Internal/Request.hs
@@ -0,0 +1,153 @@
+module Kafka.Internal.Request (
+      Produce(..)
+    , putProduceRequest
+    , putMultiProduceRequest
+
+    , Fetch(..)
+    , putFetchRequest
+    , putMultiFetchRequest
+
+    , Offsets(..)
+    , putOffsetsRequest
+    ) where
+
+import Control.Applicative
+import Data.ByteString     (ByteString)
+import Data.Foldable       (Foldable)
+
+import qualified Data.ByteString as B
+import qualified Data.Foldable   as Fold
+import qualified Data.Serialize  as C
+
+import Kafka.Internal.Types
+
+-- | 'Put's the items in the given Foldable in-order and prepends the result
+-- with a 16-bit integer containing the total number of items that were put.
+putWithCountPrefix :: Foldable f => f a -> C.Putter a -> C.Put
+putWithCountPrefix xs put =
+    let (count, p) = Fold.foldr' go (0, pure ()) xs
+    in C.putWord16be count *> p
+  where
+    go a (c, p) = (c + 1, put a *> p)
+
+-- | Prepends the given Put with a 32-bit integer containing the total size of
+-- the given Put.
+putWithLengthPrefix :: C.Put -> C.Put
+putWithLengthPrefix p = do
+    C.putWord32be $ fromIntegral (B.length bs)
+    C.putByteString bs
+  where
+    bs = C.runPut p
+
+-- | Encodes a request with the given 'RequestType'.
+--
+-- More specifically, this prepends the given put with the request type and
+-- the size of the request.
+encodeRequest :: RequestType -> C.Put -> C.Put
+encodeRequest reqType p = putWithLengthPrefix $ C.put reqType >> p
+
+-- | A request to send messages down a Kafka topic-partition pair.
+--
+-- Produce requests do not have a corresponding response. There is no way of
+-- knowing in Kafka 0.7 if a message was successfully @Produce@d.
+data Produce = Produce {
+    produceTopic     :: {-# UNPACK #-} !Topic
+  -- ^ Kafka topic to which the messages will be sent.
+  , producePartition :: {-# UNPACK #-} !Partition
+  -- ^ Partition of the topic.
+  , produceMessages  ::                [ByteString]
+  -- ^ List of message payloads.
+  --
+  -- For those concerned with low-leveld details: These messages will be
+  -- compressed using <https://code.google.com/p/snappy/ Snappy> compression.
+  } deriving (Show, Read, Eq)
+
+encodeProduce :: Produce -> C.Put
+encodeProduce (Produce topic partition messages) = do
+    C.put topic
+    C.put partition
+    putWithLengthPrefix (C.put (MessageSet messages))
+
+-- | @Put@s the given single @Produce@ request.
+putProduceRequest :: Produce -> C.Put
+putProduceRequest =
+    encodeRequest ProduceRequestType . encodeProduce
+
+-- | @Put@s the given @MultiProduce@ request.
+putMultiProduceRequest :: [Produce] -> C.Put
+putMultiProduceRequest reqs =
+    encodeRequest MultiProduceRequestType $
+        putWithCountPrefix reqs encodeProduce
+
+-- | A request to fetch messages from a particular Kafka topic-partition pair.
+--
+-- 'Kafka.FetchResponse' contains responses for this kind of request.
+data Fetch = Fetch {
+    fetchTopic     :: {-# UNPACK #-} !Topic
+  -- ^ Kafka topic from which messages will be fetched.
+  , fetchPartition :: {-# UNPACK #-} !Partition
+  -- ^ Partition of the topic.
+  , fetchOffset    :: {-# UNPACK #-} !Offset
+  -- ^ Offset at which the fetch will start.
+  --
+  -- Kafka offloads the responsiblity of knowing this to the client. That
+  -- means that if an offset is specified here that is not a real message
+  -- start, Kafka will spit out garbage.
+  --
+  -- Use 'Kafka.offsets' to find valid offsets.
+  , fetchSize      :: {-# UNPACK #-} !Size
+  -- ^ Maximum size of the returned messages.
+  --
+  -- Note, this is /not/ the number of messages. This is the maximum
+  -- combined size of the returned /compressed/ messages.
+  } deriving (Show, Read, Eq)
+
+encodeFetch :: Fetch -> C.Put
+encodeFetch (Fetch topic partition offset maxSize) = do
+    C.put topic
+    C.put partition
+    C.put offset
+    C.put maxSize
+
+-- | @Put@s the given single @Fetch@ request.
+putFetchRequest :: Fetch -> C.Put
+putFetchRequest = encodeRequest FetchRequestType . encodeFetch
+
+-- | @Put@s the given @MultiFetch@ request.
+putMultiFetchRequest :: [Fetch] -> C.Put
+putMultiFetchRequest reqs =
+    encodeRequest MultiFetchRequestType $
+        putWithCountPrefix reqs encodeFetch
+
+-- | A request to retrieve offset information from Kafka.
+--
+-- The response for this kind of request is a list of 'Offset's.
+data Offsets = Offsets {
+    offsetsTopic     :: {-# UNPACK #-} !Topic
+  -- ^ Kafka topic from which offsets will be retrieved.
+  , offsetsPartition :: {-# UNPACK #-} !Partition
+  -- ^ Partition of the topic.
+  , offsetsTime      ::                !OffsetsTime
+  -- ^ Time around which offsets will be retrieved.
+  --
+  -- If you provide a time for this, keep in mind that the response will not
+  -- contain the precise offset that occurred around that time. It will return
+  -- up to @offsetsCount@ offsets in descending, each being the first offset
+  -- of every segment file for the specified partition with a modified time
+  -- less than the specified time, and possibly a "high water mark" for the
+  -- last segment of the partition (if it was modified before the specified
+  -- time) which specifies the offset at which the next message to that
+  -- partition will be written.
+  , offsetsCount     :: {-# UNPACK #-} !Count
+  -- ^ Maximum number of offsets that will be retrieved.
+  } deriving (Show, Read, Eq)
+
+-- | @Put@s the given @Offsets@ request.
+putOffsetsRequest :: Offsets -> C.Put
+putOffsetsRequest (Offsets topic partition time maxNumber) =
+    encodeRequest OffsetsRequestType $ do
+        C.put topic
+        C.put partition
+        C.put time
+        C.put maxNumber
+
diff --git a/src/Kafka/Internal/Response.hs b/src/Kafka/Internal/Response.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Internal/Response.hs
@@ -0,0 +1,61 @@
+module Kafka.Internal.Response
+    ( Response
+    , FetchResponse(..)
+    , getFetchResponse
+    , getMultiFetchResponse
+    , getOffsetsResponse
+    ) where
+
+import Control.Applicative
+import Control.Monad
+import Data.ByteString     (ByteString)
+import Data.Monoid
+
+import qualified Data.Serialize as C
+
+import Kafka.Internal.Request
+import Kafka.Internal.Types
+
+-- | A response from Kafka can either be a failure or the value that was
+-- expected.
+type Response a = Either Error a
+
+-- | Result of a single Kafka 'Kafka.Fetch'.
+data FetchResponse = FetchResponse {
+    fetchMessages  ::                [ByteString]
+  -- ^ List of messages returned in the response for the 'Kafka.Fetch'
+  -- request.
+  , fetchNewOffset :: {-# UNPACK #-} !Offset
+  -- ^ New offset at which the next 'Fetch' request should start reading in
+  -- the same topic and partition to access the messages that follow the
+  -- messages returned in this response.
+  } deriving (Show, Read, Eq)
+
+-- | Parses a single @FetchResponse@ for the given @Fetch@ request.
+getFetchResponse :: Fetch -> C.Get FetchResponse
+getFetchResponse Fetch{fetchOffset = oldOffset} = do
+    startBytes <- C.remaining
+    messages <- fromMessageSet . mconcat <$> many C.get
+    endBytes <- C.remaining
+    let newOffset = oldOffset + fromIntegral (startBytes - endBytes)
+    return $ FetchResponse messages $! newOffset
+
+-- | Parses @FetchResponse@s for each of the given @Fetch@ requests.
+getMultiFetchResponse :: [Fetch] -> C.Get [FetchResponse]
+getMultiFetchResponse requests = forM requests $ \request -> do
+    size <- fromIntegral <$> C.getWord32be
+    C.skip 2 -- skip error
+    isolate' (size - 2) (getFetchResponse request)
+
+-- | Parses the response for an @Offsets@ request.
+getOffsetsResponse :: C.Get [Offset]
+getOffsetsResponse = do
+    numOffsets <- C.getWord32be
+    replicateM (fromIntegral numOffsets) C.get
+
+-- | A version of 'C.isolate' that does not fail the parse because of
+-- unconsumed data.
+isolate' :: Int -> C.Get a -> C.Get a
+isolate' n m = do
+    s <- C.getBytes n
+    either fail return $ C.runGet m s
diff --git a/src/Kafka/Internal/Transport.hs b/src/Kafka/Internal/Transport.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Internal/Transport.hs
@@ -0,0 +1,88 @@
+{-# LANGUAGE NamedFieldPuns #-}
+module Kafka.Internal.Transport
+    ( Transport(..)
+    , recvExactly
+
+    , Socket
+    , withConnection
+    , connect
+    , close
+    ) where
+
+
+import Control.Applicative
+import Data.ByteString     (ByteString)
+import Network.Socket      (Socket, close)
+
+import qualified Control.Exception         as E
+import qualified Data.ByteString           as B
+import qualified Data.ByteString.Char8     as B8
+import qualified Network.Socket            as N
+import qualified Network.Socket.ByteString as S
+
+-- | Types that provide a means to send and receive bytes.
+class Transport t where
+    -- | Send the given ByteString down the transport.
+    --
+    -- This must block until the request has been finished.
+    send :: t -> ByteString -> IO ()
+    -- | Read up to the given number of bytes from the stream.
+    --
+    -- The returned ByteString may be empty if the end of the stream was
+    -- reached.
+    recv :: t -> Int -> IO ByteString
+
+instance Transport Socket where
+    send = S.sendAll
+    recv = S.recv
+
+-- | Keep reading from the given transport until the given number of bytes are
+-- read or the end of the stream is reached -- whichever comes first.
+recvExactly :: Transport t => t -> Int -> IO ByteString
+recvExactly t size = B.concat . reverse <$> loop [] 0
+  where
+    loop chunks bytesRead
+        | bytesRead >= size = return chunks
+        | otherwise = do
+            chunk <- recv t (size - bytesRead)
+            if B.null chunk
+              then return chunks
+              else loop (chunk:chunks) $! bytesRead + B.length chunk
+ -- TODO this should return a bytestring of exactly the given size. Anything
+ -- else should be an IO exception or a protocol error since we use this only
+ -- in cases where we know the exact size of the response.
+
+getAddrInfo :: ByteString -> Int -> IO N.AddrInfo
+getAddrInfo host port = head <$>
+    N.getAddrInfo (Just hints)
+                  (Just $ B8.unpack host)
+                  (Just $ show port)
+  where
+    hints = N.defaultHints {
+        N.addrFamily = N.AF_INET
+      , N.addrFlags = [N.AI_NUMERICSERV]
+      , N.addrSocketType = N.Stream
+      }
+
+-- | Create a new connection.
+--
+-- Connects to the given hostname and port. Throws an 'N.IOException' in case
+-- of failure.
+connect :: ByteString -> Int -> IO Socket
+connect host port = do
+    addrInfo <- getAddrInfo host port
+    socket <- N.socket N.AF_INET N.Stream N.defaultProtocol
+    N.connect socket (N.addrAddress addrInfo) `E.onException` N.close socket
+    return socket
+
+-- | Open a connection, execute the given operation on it, and ensure it is
+-- closed afterwards even if an exception was thrown.
+--
+-- > withConnection "localhost" 9092 $ \conn ->
+-- >    doStuff conn
+-- >    fail "something went wrong"
+--
+-- Throws an 'IOException' if we were unable to open the connection.
+--
+withConnection :: ByteString -> Int -> (Socket -> IO a) -> IO a
+withConnection host port = E.bracket (connect host port) close
diff --git a/src/Kafka/Internal/Types.hs b/src/Kafka/Internal/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Internal/Types.hs
@@ -0,0 +1,289 @@
+{-# LANGUAGE FlexibleInstances          #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE OverlappingInstances       #-}
+module Kafka.Internal.Types
+    ( Error(..)
+    , Compression(..)
+    , OffsetsTime(..)
+    , RequestType(..)
+
+    , Topic(..)
+    , Offset(..)
+    , Partition(..)
+    , Size(..)
+    , Count(..)
+
+    , Message(..)
+    , MessageSet(..)
+    ) where
+
+import Control.Applicative
+import Data.ByteString       (ByteString)
+import Data.ByteString.Lazy  (fromStrict, toStrict)
+import Data.Digest.CRC32     (crc32)
+import Data.DList            (DList)
+import Data.Monoid
+import Data.String
+import Data.Time             (UTCTime)
+import Data.Time.Clock.POSIX
+import Data.Word
+
+import qualified Codec.Compression.GZip   as GZip
+import qualified Codec.Compression.Snappy as Snappy
+import qualified Data.ByteString          as B
+import qualified Data.DList               as DList
+import qualified Data.Serialize           as C
+
+-- | Different errors returned by Kafka.
+data Error
+    = UnknownError          -- -1
+    -- ^ Unknown error
+    | OffsetOutOfRangeError --  1
+    -- ^ Offset requested is invalid or no longer available on the server.
+    | InvalidMessageError   --  2
+    -- ^ A message failed to match its checksum.
+    | WrongPartitionError   --  3
+    -- ^ The requested partition doesn't exist.
+    | InvalidFetchSizeError --  4
+    -- ^ The maximum size requested for fetching is smaller than the message
+    -- being fetched.
+  deriving (Show, Read, Eq, Ord)
+
+instance C.Serialize (Maybe Error) where
+    -- A note about extensions:
+    -- We need FlexibleInstances to allow GHC to let us declare an instance
+    -- for the type @Maybe Error@. We also need OverlappingInstances to tell
+    -- it to pick this (more specific) instance of Maybe Error over the
+    -- generic one provided by cereal.
+    get = do
+        code <- C.getWord16be
+        case code of
+            0  -> return Nothing
+            1  -> return $ Just OffsetOutOfRangeError
+            2  -> return $ Just InvalidMessageError
+            3  -> return $ Just WrongPartitionError
+            4  -> return $ Just InvalidFetchSizeError
+            -1 -> return $ Just UnknownError
+            _  ->   fail $ "Unknown error code: " ++ show code
+
+    put                     Nothing  = C.putWord16be 0
+    put (Just OffsetOutOfRangeError) = C.putWord16be 1
+    put (Just   InvalidMessageError) = C.putWord16be 2
+    put (Just   WrongPartitionError) = C.putWord16be 3
+    put (Just InvalidFetchSizeError) = C.putWord16be 4
+    put (Just          UnknownError) = C.putWord16be (-1)
+
+-- | Methods of compression supported by Kafka.
+data Compression
+    = NoCompression
+    -- ^ The message is uncompressed.
+    | GzipCompression
+    -- ^ The message is compressed using @gzip@ compression and may contain
+    -- other messages in it.
+    | SnappyCompression
+    -- ^ The message is compressed using @snappy@ compression and may contain
+    -- other messages in it.
+  deriving (Show, Read, Eq, Ord)
+
+instance C.Serialize Compression where
+    put NoCompression     = C.putWord8 0
+    put GzipCompression   = C.putWord8 1
+    put SnappyCompression = C.putWord8 2
+
+    get = C.getWord8 >>= \i -> case i of
+        0 -> return NoCompression
+        1 -> return GzipCompression
+        2 -> return SnappyCompression
+        _ -> fail $ "Invalid compression code: " ++ show i
+
+-- | Different times for which offsets may be retrieved using
+-- 'Kafka.offsets'.
+data OffsetsTime
+    = OffsetsLatest
+    -- ^ Retrieve the latest offsets
+    | OffsetsEarliest
+    -- ^ Retrieve the earliest offsets.
+    | OffsetsBefore !UTCTime
+    -- ^ Retrieve offsets before the given time.
+    --
+    -- Keep in mind that the response will not contain the precise offset that
+    -- occurred around this time. It will return up to the specified count of
+    -- offsets in descending, each being the first offset of every segment
+    -- file for the specified partition with a modified time less than this
+    -- time, and possibly a "high water mark" for the last segment of the
+    -- partition (if it was modified before this time) which specifies the
+    -- offset at which the next message to that partition will be written.
+  deriving (Show, Read, Eq, Ord)
+
+instance C.Serialize OffsetsTime where
+    put OffsetsLatest = C.putWord64be (-1)
+    put OffsetsEarliest = C.putWord64be (-2)
+    put (OffsetsBefore t) =
+        C.putWord64be . round . (* 1000) . utcTimeToPOSIXSeconds $ t
+
+    -- Kafka doesn't care about anything beyond milliseconds while UTCTime can
+    -- be more precise. We will lose some amount of precision if the same
+    -- UTCTime is serialized and then deserialized.
+    get = C.getWord64be >>= \i -> case i of
+        -1 -> return   OffsetsLatest
+        -2 -> return   OffsetsEarliest
+        t  -> return . OffsetsBefore .
+               posixSecondsToUTCTime . (/ 1000) . realToFrac $ t
+
+-- | The different request types supported by Kafka.
+data RequestType
+    = ProduceRequestType        -- 0
+    | FetchRequestType          -- 1
+    | MultiFetchRequestType     -- 2
+    | MultiProduceRequestType   -- 3
+    | OffsetsRequestType        -- 4
+  deriving (Show, Read, Eq, Ord)
+
+instance C.Serialize RequestType where
+    put ProduceRequestType      = C.putWord16be 0
+    put FetchRequestType        = C.putWord16be 1
+    put MultiFetchRequestType   = C.putWord16be 2
+    put MultiProduceRequestType = C.putWord16be 3
+    put OffsetsRequestType      = C.putWord16be 4
+
+    get = C.getWord16be >>= \i -> case i of
+            0 -> return ProduceRequestType
+            1 -> return FetchRequestType
+            2 -> return MultiFetchRequestType
+            3 -> return MultiProduceRequestType
+            4 -> return OffsetsRequestType
+            _ -> fail $ "Unknown request type: " ++ show i
+
+-- | Represents a Kafka topic.
+--
+-- This is an instance of 'IsString' so a literal string may be used to create
+-- a Topic with the @OverloadedStrings@ extension.
+newtype Topic = Topic ByteString
+    deriving (Show, Read, Eq, Ord, IsString)
+
+instance C.Serialize Topic where
+    put (Topic topic) = do
+        C.putWord16be (fromIntegral $ B.length topic)
+        C.putByteString topic
+    get = do
+        topicLength <- fromIntegral <$> C.getWord16be
+        Topic <$> C.getByteString topicLength
+
+-- | Represents an Offset in Kafka.
+--
+-- This is an instance of 'Num' so a literal number may be used to create an
+-- Offset.
+newtype Offset = Offset Word64
+    deriving (Show, Read, Eq, Ord, Num)
+
+instance C.Serialize Offset where
+    put (Offset o) = C.putWord64be o
+    get = Offset <$> C.getWord64be
+
+-- | Represents a Kafka topic partition.
+--
+-- This is an instance of 'Num' so a literal number may be used to create a
+-- Partition.
+newtype Partition = Partition Word32
+    deriving (Show, Read, Eq, Ord, Num)
+
+instance C.Serialize Partition where
+    put (Partition p) = C.putWord32be p
+    get = Partition <$> C.getWord32be
+
+-- | Represents a size.
+--
+-- This is an instance of 'Num' so a literal number may be used to create a
+-- Size.
+newtype Size = Size Word32
+    deriving (Show, Read, Eq, Ord, Num)
+
+instance C.Serialize Size where
+    put (Size s) = C.putWord32be s
+    get = Size <$> C.getWord32be
+
+-- | Represents a Count.
+--
+-- This is an instance of 'Num' so a literal number may be used to create a
+-- Size.
+newtype Count = Count Word32
+    deriving (Show, Read, Eq, Ord, Num)
+
+instance C.Serialize Count where
+    put (Count s) = C.putWord32be s
+    get = Count <$> C.getWord32be
+
+-- | Represents a Message being sent through Kafka.
+data Message = Message {
+    messageCompression :: !Compression
+  -- ^ Compression used for the message.
+  --
+  -- If this is anything but 'NoCompression', this message contains other
+  -- messages in it.
+  , messagePayload     :: !ByteString
+  -- ^ Message payload.
+  --
+  -- If the message is using any compression scheme, the payload contains
+  -- other messages in the same format.
+  } deriving (Show, Read, Eq, Ord)
+
+instance C.Serialize Message where
+    put (Message compression payload) = do
+        C.putWord32be (fromIntegral messageLength)
+        C.putWord8 1 -- Magic
+        C.put compression
+        C.putWord32be (crc32 payload)
+        C.putByteString payload
+      where
+        messageLength = B.length payload
+            + 4 -- checksum
+            + 1 -- compression
+            + 1 -- magic
+
+    get = C.getWord32be >>= \messageLength -> do
+        magic <- C.getWord8
+        compression <- case magic of
+            0 -> return NoCompression
+            1 -> C.get
+            _ -> fail $ "Unknown magic code: " ++ show magic
+        checksum <- C.getWord32be
+        let remainingLength = fromIntegral $
+              messageLength -
+                ( 4                  -- checksum
+                + 1                  -- magic
+                + fromIntegral magic -- compression (1 byte if magic is 1)
+                )
+        payload <- C.getByteString remainingLength
+        if crc32 payload == checksum
+          then return (Message compression payload)
+          else fail "Checksum did not match."
+            -- TODO: This should probably be InvalidMessageError
+
+-- | Represents a collection of message payloads.
+--
+-- These are compressed into a single message using Snappy compression when
+-- being sent.
+newtype MessageSet = MessageSet { fromMessageSet :: [ByteString] }
+    deriving (Show, Read, Eq, Monoid)
+
+instance C.Serialize MessageSet where
+    put (MessageSet messages) = C.put (Message SnappyCompression payload)
+      where
+        payload = Snappy.compress . C.runPut $
+                    mapM_ (C.put . Message NoCompression) messages
+
+    get = MessageSet . DList.toList <$> (C.get >>= readMessages)
+      where
+        readMessages :: Message -> C.Get (DList ByteString)
+        readMessages (Message NoCompression payload) =
+            return (DList.singleton payload)
+        readMessages (Message compression payload) = do
+            messages <- either fail return $
+                        C.runGet (many C.get) decompressedPayload
+            mconcat <$> mapM readMessages messages
+          where
+            decompressedPayload = decompress payload
+            decompress = case compression of
+              NoCompression -> id
+              SnappyCompression -> Snappy.decompress
+              GzipCompression -> toStrict . GZip.decompress . fromStrict
diff --git a/test-lib/Test/Kafka/Arbitrary.hs b/test-lib/Test/Kafka/Arbitrary.hs
new file mode 100644
--- /dev/null
+++ b/test-lib/Test/Kafka/Arbitrary.hs
@@ -0,0 +1,87 @@
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE StandaloneDeriving         #-}
+module Test.Kafka.Arbitrary
+    ( Payload(..)
+    , NonEmptyPayloadList(..)
+    ) where
+
+import Control.Applicative
+import Test.QuickCheck
+
+import qualified Data.ByteString       as B
+import qualified Data.ByteString.Char8 as B8
+import qualified Data.Time             as T
+
+import qualified Kafka.Internal.Types as I
+
+instance Arbitrary I.Error where
+    arbitrary = elements [
+        I.UnknownError
+      , I.OffsetOutOfRangeError
+      , I.InvalidMessageError
+      , I.WrongPartitionError
+      , I.InvalidFetchSizeError
+      ]
+
+instance Arbitrary I.Compression where
+    arbitrary = elements [
+        I.NoCompression
+      , I.SnappyCompression
+      , I.GzipCompression
+      ]
+
+instance Arbitrary I.OffsetsTime where
+    arbitrary = oneof [
+        elements [I.OffsetsLatest, I.OffsetsEarliest]
+      , I.OffsetsBefore <$> someTime
+      ]
+      where
+        someTime = T.UTCTime <$> someDay <*> someDayTime
+        someDay =
+            -- Note: We only support times after Unix epoch. Any dates before
+            -- 1970 cannot be represented in Kafka 0.7.
+            T.fromGregorian <$> choose (1970, 2400)
+                            <*> choose (1, 12)
+                            <*> choose (0, 31)
+        someDayTime = T.secondsToDiffTime <$> choose (0, 86400)
+
+instance Arbitrary I.RequestType where
+    arbitrary = elements [
+        I.ProduceRequestType
+      , I.FetchRequestType
+      , I.MultiFetchRequestType
+      , I.MultiProduceRequestType
+      , I.OffsetsRequestType
+      ]
+
+instance Arbitrary I.Topic where
+    arbitrary = I.Topic . B8.pack <$> safeString
+      where
+        safeString = listOf1 $ elements ['a'..'z']
+
+deriving instance Arbitrary I.Offset
+deriving instance Arbitrary I.Partition
+deriving instance Arbitrary I.Size
+deriving instance Arbitrary I.Count
+
+instance Arbitrary I.Message where
+    arbitrary =
+        I.Message <$> arbitrary
+                  <*> (getPayload <$> arbitrary)
+
+instance Arbitrary I.MessageSet where
+    arbitrary = I.MessageSet . map getPayload <$> arbitrary
+
+newtype Payload = Payload { getPayload :: B.ByteString }
+    deriving (Show)
+
+instance Arbitrary Payload where
+    arbitrary = Payload . B.pack <$> arbitrary
+
+newtype NonEmptyPayloadList = NonEmptyPayloadList
+    { getPayloads :: [B.ByteString]
+    } deriving (Show)
+
+instance Arbitrary NonEmptyPayloadList where
+    arbitrary = NonEmptyPayloadList . map getPayload <$> listOf1 arbitrary
diff --git a/test-lib/Test/Kafka/Managed.hs b/test-lib/Test/Kafka/Managed.hs
new file mode 100644
--- /dev/null
+++ b/test-lib/Test/Kafka/Managed.hs
@@ -0,0 +1,96 @@
+{-# LANGUAGE RecordWildCards #-}
+module Test.Kafka.Managed
+    ( KafkaConfig(..)
+    , withManagedKafka
+    ) where
+
+import Control.Applicative
+import Control.Concurrent  (threadDelay)
+import Control.Exception   (IOException, bracket, catch)
+import System.Exit         (ExitCode (..))
+
+import qualified Network.Socket as N
+import qualified System.IO      as IO
+import qualified System.IO.Temp as Temp
+import qualified System.Process as P
+
+data KafkaConfig = KafkaConfig {
+    kafkaServerPort   :: Int
+  , kafkaLogDirectory :: FilePath
+  } deriving (Show, Eq)
+
+toProperties :: KafkaConfig -> [(String, String)]
+toProperties KafkaConfig{..} =
+    [ ("brokerid", "0")
+    , ("port", show kafkaServerPort)
+    , ("log.dir", kafkaLogDirectory)
+    , ("num.partitions", "1")
+    , ("enable.zookeeper", "false")
+    , ("log.flush.interval", "1")
+    , ("log.default.flush.interval.ms", "10")
+    , ("log.default.flush.scheduler.interval.ms", "5")
+    ]
+
+writeProperties :: [(String, String)] -> IO.Handle -> IO ()
+writeProperties entries h = IO.hPutStr h . unlines $ do
+    (key, value) <- entries
+    return $! key ++ "=" ++ value
+
+withManagedKafka :: KafkaConfig -> IO a -> IO a
+withManagedKafka config m =
+  Temp.withSystemTempFile "kafka.proeprties" $ \configPath handle -> do
+    writeProperties (toProperties config) handle
+    IO.hClose handle
+    bracket (startServer configPath) stopServer . const $ do
+        waitUntilConnectable (kafkaServerPort config)
+        m
+
+-- | Keep trying to connect to the given port until the server accepts a
+-- connection or we've made more than 10 attempts.
+waitUntilConnectable :: Int -> IO ()
+waitUntilConnectable port = loop (0 :: Int)
+  where
+    loop attempts
+        -- TODO maxAttempts can probably be a parameter
+      | attempts >= 10 = error "Can't connect to Kafka after 10 attempts."
+      | otherwise = do
+          addrInfo <- head <$> N.getAddrInfo
+                  (Just $ N.defaultHints {
+                          N.addrFlags = [N.AI_NUMERICSERV]
+                        , N.addrFamily = N.AF_INET
+                        , N.addrSocketType = N.Stream
+                        })
+                  (Just "localhost") (Just $ show port)
+          socket <- N.socket N.AF_INET N.Stream N.defaultProtocol
+          let connect = do
+                  N.connect socket (N.addrAddress addrInfo)
+                  N.close socket
+              retry = do
+                  threadDelay (200 * 1000)
+                  loop (attempts + 1)
+          connect `catchIO` const retry
+    catchIO :: IO a -> (IOException -> IO a) -> IO a
+    catchIO = catch
+
+-- TODO Better stdout/stderr handling. Could return a ManagedKafka object here
+-- that allows access to the logs in case of failure.
+startServer :: FilePath -> IO P.ProcessHandle
+startServer configPath =
+  IO.withFile "/dev/null" IO.WriteMode $ \nullHandle -> do
+    (_, _, _, h) <- P.createProcess (p nullHandle)
+    return h
+  where
+    p nullHandle = (P.proc "kafka-server-start.sh" [configPath]) {
+        P.std_out = P.UseHandle nullHandle
+      }
+
+stopServer :: P.ProcessHandle -> IO ()
+stopServer h = do
+    P.interruptProcessGroupOf h
+    code <- P.waitForProcess h
+    case code of
+      ExitSuccess -> return ()
+      ExitFailure 130 -> return ()  -- Exit code 130 == terminated with Ctrl-C
+      ExitFailure exitCode -> putStrLn . unwords $
+          ["Failed to stop Kafka server. Exit code", show exitCode]
+
diff --git a/test/Kafka/Internal/TypesSpec.hs b/test/Kafka/Internal/TypesSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Kafka/Internal/TypesSpec.hs
@@ -0,0 +1,61 @@
+module Kafka.Internal.TypesSpec where
+
+import Test.Hspec
+import Test.QuickCheck
+
+import qualified Data.Serialize as C
+
+import Test.Kafka.Arbitrary ()
+
+import qualified Kafka.Internal.Types as I
+
+spec :: Spec
+spec = do
+    describe "Error" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: Maybe I.Error -> Expectation)
+
+    describe "Compression" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Compression -> Expectation)
+
+    describe "OffsetsTime" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.OffsetsTime -> Expectation)
+
+    describe "RequestType" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.RequestType -> Expectation)
+
+    describe "Topic" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Topic -> Expectation)
+
+    describe "Offset" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Offset -> Expectation)
+
+    describe "Partition" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Partition -> Expectation)
+
+    describe "Size" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Size -> Expectation)
+
+    describe "Count" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Count -> Expectation)
+
+    describe "Message" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.Message -> Expectation)
+
+    describe "MessageSet" $
+        it "serializes and deserializes" $
+            property (checkSerialization :: I.MessageSet -> Expectation)
+
+-- | Check that the given serializable item serializes correctly in both
+-- directions.
+checkSerialization :: (C.Serialize a, Show a, Eq a) => a -> Expectation
+checkSerialization a = C.decode (C.encode a) `shouldBe` Right a