5 files changed, 1027 insertions, 0 deletions

diff --git a/src/Data/BEncode/Pretty.hs b/src/Data/BEncode/Pretty.hs
new file mode 100644
index 00000000..7b0d46a0
--- /dev/null
+++ b/src/Data/BEncode/Pretty.hs
@@ -0,0 +1,75 @@
+{-# LANGUAGE CPP                #-}
+module Data.BEncode.Pretty where -- (showBEncode) where
+
+import Data.BEncode.Types
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy.Char8 as BL8
+import Data.Text (Text)
+import qualified Data.Text as T
+#ifdef BENCODE_AESON
+import Data.BEncode.BDict hiding (map)
+import Data.Aeson.Types hiding (parse)
+import Data.Aeson.Encode.Pretty
+import qualified Data.HashMap.Strict as HashMap
+import qualified Data.Vector as Vector
+import Data.Foldable as Foldable
+import Data.Text.Encoding
+import Text.Printf
+#endif
+
+#ifdef BENCODE_AESON
+
+unhex :: Text -> BS.ByteString
+unhex t = BS.pack $ map unhex1 [0 .. BS.length nibs `div` 2]
+ where
+    nibs = encodeUtf8 t
+    unhex1 i = unnib (BS.index nibs (i * 2)) * 0x10
+                + unnib (BS.index nibs (i * 2 + 1))
+    unnib a | a <= 0x39 = a - 0x30
+            | otherwise = a - (0x41 - 10)
+
+hex :: BS.ByteString -> Text
+hex bs = T.concat $ map (T.pack . printf "%02X") $ BS.unpack bs
+
+quote_chr :: Char
+quote_chr = ' '
+
+quote :: Text -> Text
+quote t = quote_chr `T.cons` t `T.snoc` quote_chr
+
+
+instance ToJSON BValue where
+    toJSON (BInteger x) = Number $ fromIntegral x
+    toJSON (BString s) = String $ either (const $ hex s) quote $ decodeUtf8' s
+    toJSON (BList xs) = Array $ Vector.fromList $ map toJSON xs
+    toJSON (BDict d) = toJSON d
+
+instance ToJSON a => ToJSON (BDictMap a) where
+    toJSON d = Object $ HashMap.fromList $ map convert $ toAscList d
+       where
+        convert (k,v) = (decodeUtf8 k,toJSON v)
+
+instance FromJSON BValue where
+    parseJSON (Number x) = pure $ BInteger (truncate x)
+    parseJSON (Bool x)   = pure $ BInteger $ if x then 1 else 0
+    parseJSON (String s)
+        | T.head s==quote_chr = pure $ BString $ encodeUtf8 (T.takeWhile (/=quote_chr) $ T.drop 1 s)
+        | otherwise           = pure $ BString $ unhex s
+    parseJSON (Array v)  = BList <$> traverse parseJSON (Foldable.toList v)
+    parseJSON (Object d) = BDict <$> parseJSON (Object d)
+    parseJSON (Null)     = pure $ BDict Nil
+
+instance FromJSON v => FromJSON (BDictMap v) where
+    parseJSON (Object d) = fromAscList <$> traverse convert (HashMap.toList d)
+        where
+            convert (k,v) = (,) (encodeUtf8 k) <$> parseJSON v
+    parseJSON _ = fail "Not a BDict"
+#endif
+
+showBEncode :: BValue -> BL.ByteString
+#ifdef BENCODE_AESON
+showBEncode b = encodePretty $ toJSON b
+#else
+showBEncode b = BL8.pack (show b)
+#endif
diff --git a/src/Network/KRPC.hs b/src/Network/KRPC.hs
new file mode 100644
index 00000000..d185fb4c
--- /dev/null
+++ b/src/Network/KRPC.hs
@@ -0,0 +1,91 @@
+-- |
+--   Copyright   :  (c) Sam Truzjan 2013, 2014
+--   License     :  BSD3
+--   Maintainer  :  pxqr.sta@gmail.com
+--   Stability   :  experimental
+--   Portability :  portable
+--
+--   This module provides safe remote procedure call. One important
+--   point is exceptions and errors, so to be able handle them
+--   properly we need to investigate a bit about how this all works.
+--   Internally, in order to make method invokation KRPC makes the
+--   following steps:
+--
+--     * Caller serialize arguments to bencoded bytestrings;
+--
+--     * Caller send bytestring data over UDP to the callee;
+--
+--     * Callee receive and decode arguments to the method and method
+--     name. If it can't decode then it send 'ProtocolError' back to the
+--     caller;
+--
+--     * Callee search for the @method name@ in the method table.
+--     If it not present in the table then callee send 'MethodUnknown'
+--     back to the caller;
+--
+--     * Callee check if argument names match. If not it send
+--     'ProtocolError' back;
+--
+--     * Callee make the actuall call to the plain old haskell
+--       function.  If the function throw exception then callee send
+--       'ServerError' back.
+--
+--     * Callee serialize result of the function to bencoded bytestring.
+--
+--     * Callee encode result to bencoded bytestring and send it back
+--     to the caller.
+--
+--     * Caller check if return values names match with the signature
+--     it called in the first step.
+--
+--     * Caller extracts results and finally return results of the
+--     procedure call as ordinary haskell values.
+--
+--   If every other error occurred then caller get the
+--   'GenericError'. All errors returned by callee are throwed as
+--   ordinary haskell exceptions at caller side. Also note that both
+--   caller and callee use plain UDP, so KRPC is unreliable.
+--
+--   For async 'query' use @async@ package.
+--
+--   For protocol details see "Network.KRPC.Message" module.
+--
+module Network.KRPC
+       ( -- * Methods
+         Method
+       , KRPC (..)
+
+         -- * RPC
+         -- ** Query
+       , QueryFailure (..)
+       , query
+       , query'
+       , queryRaw
+       , getQueryCount
+
+         -- ** Handler
+       , HandlerFailure (..)
+       , Handler
+       , handler
+
+         -- * Manager
+       , MonadKRPC (..)
+       , Options (..)
+       , def
+       , Manager
+       , newManager
+       , closeManager
+       , withManager
+       , isActive
+       , listen
+
+         -- * Re-exports
+       , ErrorCode (..)
+       , SockAddr (..)
+       ) where
+
+import Data.Default.Class
+import Network.KRPC.Message
+import Network.KRPC.Method
+import Network.KRPC.Manager
+import Network.Socket (SockAddr (..))
diff --git a/src/Network/KRPC/Manager.hs b/src/Network/KRPC/Manager.hs
new file mode 100644
index 00000000..c90c92f9
--- /dev/null
+++ b/src/Network/KRPC/Manager.hs
@@ -0,0 +1,485 @@
+-- |
+--   Copyright   :  (c) Sam Truzjan 2013, 2014
+--   License     :  BSD3
+--   Maintainer  :  pxqr.sta@gmail.com
+--   Stability   :  experimental
+--   Portability :  portable
+--
+--   Normally, you don't need to import this module.
+--
+{-# LANGUAGE OverloadedStrings      #-}
+{-# LANGUAGE FlexibleInstances      #-}
+{-# LANGUAGE FlexibleContexts       #-}
+{-# LANGUAGE ScopedTypeVariables    #-}
+{-# LANGUAGE DefaultSignatures      #-}
+{-# LANGUAGE MultiParamTypeClasses  #-}
+{-# LANGUAGE FunctionalDependencies #-}
+{-# LANGUAGE DeriveDataTypeable     #-}
+{-# LANGUAGE TemplateHaskell        #-}
+module Network.KRPC.Manager
+       ( -- * Manager
+         MonadKRPC (..)
+       , Options (..)
+       , Manager
+       , newManager
+       , closeManager
+       , withManager
+       , isActive
+       , listen
+
+         -- * Queries
+       , QueryFailure (..)
+       , query
+       , query'
+       , queryRaw
+       , getQueryCount
+
+         -- * Handlers
+       , HandlerFailure (..)
+       , Handler
+       , handler
+       ) where
+
+import Control.Applicative
+import Control.Concurrent
+import Control.Concurrent.Lifted (fork)
+import Control.Exception hiding (Handler)
+import qualified Control.Exception.Lifted as E (Handler (..))
+import Control.Exception.Lifted as Lifted (catches, finally)
+import Control.Monad
+import Control.Monad.Logger
+import Control.Monad.Reader
+import Control.Monad.Trans.Control
+import Data.BEncode as BE
+import Data.BEncode.Internal as BE
+import Data.BEncode.Pretty (showBEncode)
+import Data.ByteString as BS
+import Data.ByteString.Char8 as BC
+import Data.ByteString.Lazy as BL
+import Data.Default.Class
+import Data.IORef
+import Data.List as L
+import Data.Map as M
+import Data.Monoid
+import Data.Text as T
+import Data.Text.Encoding as T
+import Data.Tuple
+import Data.Typeable
+import Network.KRPC.Message
+import Network.KRPC.Method
+import Network.Socket hiding (listen)
+import Network.Socket.ByteString as BS
+import System.IO.Error
+import System.Timeout
+
+
+{-----------------------------------------------------------------------
+--  Options
+-----------------------------------------------------------------------}
+
+-- | RPC manager options.
+data Options = Options
+  { -- | Initial 'TransactionId' incremented with each 'query';
+    optSeedTransaction :: {-# UNPACK #-} !Int
+
+    -- | Time to wait for response from remote node, in seconds.
+  , optQueryTimeout    :: {-# UNPACK #-} !Int
+
+    -- | Maximum number of bytes to receive.
+  , optMaxMsgSize      :: {-# UNPACK #-} !Int
+  } deriving (Show, Eq)
+
+defaultSeedTransaction :: Int
+defaultSeedTransaction = 0
+
+defaultQueryTimeout :: Int
+defaultQueryTimeout = 120
+
+defaultMaxMsgSize :: Int
+defaultMaxMsgSize = 64 * 1024
+
+-- | Permissive defaults.
+instance Default Options where
+  def = Options
+    { optSeedTransaction = defaultSeedTransaction
+    , optQueryTimeout    = defaultQueryTimeout
+    , optMaxMsgSize      = defaultMaxMsgSize
+    }
+
+validateOptions :: Options -> IO ()
+validateOptions Options {..}
+  | optQueryTimeout < 1
+  = throwIO (userError "krpc: non-positive query timeout")
+  | optMaxMsgSize   < 1
+  = throwIO (userError "krpc: non-positive buffer size")
+  |       otherwise     = return ()
+
+{-----------------------------------------------------------------------
+--  Options
+-----------------------------------------------------------------------}
+
+type KResult = Either KError KResponse
+
+type TransactionCounter = IORef Int
+type CallId             = (TransactionId, SockAddr)
+type CallRes            = MVar (BValue, KResult)
+type PendingCalls       = IORef (Map CallId CallRes)
+
+type HandlerBody h = SockAddr -> BValue -> h (BE.Result BValue)
+
+-- | Handler is a function which will be invoked then some /remote/
+-- node querying /this/ node.
+type Handler     h = (MethodName, HandlerBody h)
+
+-- | Keep track pending queries made by /this/ node and handle queries
+-- made by /remote/ nodes.
+data Manager h = Manager
+  { sock               :: !Socket
+  , options            :: !Options
+  , listenerThread     :: !(MVar ThreadId)
+  , transactionCounter :: {-# UNPACK #-} !TransactionCounter
+  , pendingCalls       :: {-# UNPACK #-} !PendingCalls
+  , handlers           :: [Handler h]
+  }
+
+-- | A monad which can perform or handle queries.
+class (MonadBaseControl IO m, MonadLogger m, MonadIO m)
+    => MonadKRPC h m | m -> h where
+
+  -- | Ask for manager.
+  getManager :: m (Manager h)
+
+  default getManager :: MonadReader (Manager h) m => m (Manager h)
+  getManager = ask
+
+  -- | Can be used to add logging for instance.
+  liftHandler :: h a -> m a
+
+  default liftHandler :: m a -> m a
+  liftHandler = id
+
+instance (MonadBaseControl IO h, MonadLogger h, MonadIO h)
+      => MonadKRPC h (ReaderT (Manager h) h) where
+
+  liftHandler = lift
+
+sockAddrFamily :: SockAddr -> Family
+sockAddrFamily (SockAddrInet  _ _    ) = AF_INET
+sockAddrFamily (SockAddrInet6 _ _ _ _) = AF_INET6
+sockAddrFamily (SockAddrUnix  _      ) = AF_UNIX
+sockAddrFamily (SockAddrCan _        ) = AF_CAN
+
+-- | Bind socket to the specified address. To enable query handling
+-- run 'listen'.
+newManager :: Options        -- ^ various protocol options;
+           -> SockAddr       -- ^ address to listen on;
+           -> [Handler h]    -- ^ handlers to run on incoming queries.
+           -> IO (Manager h) -- ^ new rpc manager.
+newManager opts @ Options {..} servAddr handlers = do
+    validateOptions opts
+    sock  <- bindServ
+    tref  <- newEmptyMVar
+    tran  <- newIORef optSeedTransaction
+    calls <- newIORef M.empty
+    return $ Manager sock opts tref tran calls handlers
+  where
+    bindServ = do
+      let family = sockAddrFamily servAddr
+      sock <- socket family Datagram defaultProtocol
+      when (family == AF_INET6) $ do
+        setSocketOption sock IPv6Only 0
+      bindSocket sock servAddr
+      return sock
+
+-- | Unblock all pending calls and close socket.
+closeManager :: Manager m -> IO ()
+closeManager Manager {..} = do
+  maybe (return ()) killThread =<< tryTakeMVar listenerThread
+  -- TODO unblock calls
+  close sock
+
+-- | Check if the manager is still active. Manager becomes active
+-- until 'closeManager' called.
+isActive :: Manager m -> IO Bool
+isActive Manager {..} = liftIO $ isBound sock
+{-# INLINE isActive #-}
+
+-- | Normally you should use Control.Monad.Trans.Resource.allocate
+-- function.
+withManager :: Options -> SockAddr -> [Handler h]
+            -> (Manager h -> IO a) -> IO a
+withManager opts addr hs = bracket (newManager opts addr hs) closeManager
+
+{-----------------------------------------------------------------------
+--  Logging
+-----------------------------------------------------------------------}
+
+-- TODO prettify log messages
+querySignature :: MethodName -> TransactionId -> SockAddr -> Text
+querySignature name transaction addr = T.concat
+  [  "&", T.decodeUtf8 name
+  , " #", T.decodeUtf8 transaction
+  , " @", T.pack (show addr)
+  ]
+
+{-----------------------------------------------------------------------
+--  Client
+-----------------------------------------------------------------------}
+-- we don't need to know about TransactionId while performing query,
+-- so we introduce QueryFailure exceptions
+
+-- | Used to signal 'query' errors.
+data QueryFailure
+  = SendFailed                  -- ^ unable to send query;
+  | QueryFailed  ErrorCode Text -- ^ remote node return error;
+  | TimeoutExpired              -- ^ remote node not responding.
+  deriving (Show, Eq, Typeable)
+
+instance Exception QueryFailure
+
+sendMessage :: MonadIO m => BEncode a => Socket -> SockAddr -> a -> m ()
+sendMessage sock addr a = do
+  liftIO $ sendManyTo sock (BL.toChunks (BE.encode a)) addr
+
+genTransactionId :: TransactionCounter -> IO TransactionId
+genTransactionId ref = do
+  cur <- atomicModifyIORef' ref $ \ cur -> (succ cur, cur)
+  return $ BC.pack (show cur)
+
+-- | How many times 'query' call have been performed.
+getQueryCount :: MonadKRPC h m => m Int
+getQueryCount = do
+  Manager {..} <- getManager
+  curTrans <- liftIO $ readIORef transactionCounter
+  return $ curTrans - optSeedTransaction options
+
+registerQuery :: CallId -> PendingCalls -> IO CallRes
+registerQuery cid ref = do
+  ares <- newEmptyMVar
+  atomicModifyIORef' ref $ \ m -> (M.insert cid ares m, ())
+  return ares
+
+-- simultaneous M.lookup and M.delete guarantees that we never get two
+-- or more responses to the same query
+unregisterQuery :: CallId -> PendingCalls -> IO (Maybe CallRes)
+unregisterQuery cid ref = do
+  atomicModifyIORef' ref $ swap .
+    M.updateLookupWithKey (const (const Nothing)) cid
+
+
+-- (sendmsg EINVAL)
+sendQuery :: BEncode a => Socket -> SockAddr -> a -> IO ()
+sendQuery sock addr q = handle sockError $ sendMessage sock addr q
+  where
+    sockError :: IOError -> IO ()
+    sockError _ = throwIO SendFailed
+
+-- | Enqueue query to the given node.
+--
+--  This function should throw 'QueryFailure' exception if quered node
+--  respond with @error@ message or the query timeout expires.
+--
+query :: forall h m a b. (MonadKRPC h m, KRPC a b) => SockAddr -> a -> m b
+query addr params = queryK addr params (\_ x _ -> x)
+
+-- | Like 'query' but possibly returns your externally routable IP address.
+query' :: forall h m a b. (MonadKRPC h m, KRPC a b) => SockAddr -> a -> m (b, Maybe ReflectedIP)
+query' addr params = queryK addr params (const (,))
+
+-- | Enqueue a query, but give us the complete BEncoded content sent by the
+-- remote Node.  This is useful for handling extensions that this library does
+-- not otherwise support.
+queryRaw :: forall h m a b. (MonadKRPC h m, KRPC a b) => SockAddr -> a -> m (b, BValue)
+queryRaw addr params = queryK addr params (\raw x _ -> (x,raw))
+
+queryK :: forall h m a b x. (MonadKRPC h m, KRPC a b) =>
+    SockAddr -> a -> (BValue -> b -> Maybe ReflectedIP -> x) -> m x
+queryK addr params kont = do
+  Manager {..} <- getManager
+  tid <- liftIO $ genTransactionId transactionCounter
+  let queryMethod = method :: Method a b
+  let signature = querySignature (methodName queryMethod) tid addr
+  $(logDebugS) "query.sending" signature
+
+  mres <- liftIO $ do
+    ares <- registerQuery (tid, addr) pendingCalls
+
+    let q = KQuery (toBEncode params) (methodName queryMethod) tid
+    sendQuery sock addr q
+      `onException` unregisterQuery (tid, addr) pendingCalls
+
+    timeout (optQueryTimeout options * 10 ^ (6 :: Int)) $ do
+      (raw,res) <- readMVar ares
+      case res of
+        Left  (KError   c m _) -> throwIO $ QueryFailed c (T.decodeUtf8 m)
+        Right (KResponse {..}) ->
+          case fromBEncode respVals of
+            Right r -> pure $ kont raw r respIP
+            Left  e -> throwIO $ QueryFailed ProtocolError (T.pack e)
+
+  case mres of
+    Just res -> do
+      $(logDebugS) "query.responded" $ signature
+      return res
+
+    Nothing -> do
+      _ <- liftIO $ unregisterQuery (tid, addr) pendingCalls
+      $(logWarnS) "query.not_responding" $ signature <> " for " <>
+             T.pack (show (optQueryTimeout options)) <> " seconds"
+      throw $ TimeoutExpired
+
+{-----------------------------------------------------------------------
+--  Handlers
+-----------------------------------------------------------------------}
+-- we already throw:
+--
+--   * ErrorCode(MethodUnknown) in the 'dispatchHandler';
+--
+--   * ErrorCode(ServerError) in the 'runHandler';
+--
+--   * ErrorCode(GenericError) in the 'runHandler' (those can be
+--   async exception too)
+--
+--  so HandlerFailure should cover *only* 'ProtocolError's.
+
+-- | Used to signal protocol errors.
+data HandlerFailure
+  = BadAddress            -- ^ for e.g.: node calls herself;
+  | InvalidParameter Text -- ^ for e.g.: bad session token.
+    deriving (Show, Eq, Typeable)
+
+instance Exception HandlerFailure
+
+prettyHF :: HandlerFailure -> BS.ByteString
+prettyHF  BadAddress               = T.encodeUtf8 "bad address"
+prettyHF (InvalidParameter reason) = T.encodeUtf8 $
+  "invalid parameter: " <> reason
+
+prettyQF :: QueryFailure -> BS.ByteString
+prettyQF e = T.encodeUtf8 $ "handler fail while performing query: "
+  <> T.pack (show e)
+
+-- | Make handler from handler function. Any thrown exception will be
+-- supressed and send over the wire back to the querying node.
+--
+--   If the handler make some 'query' normally it /should/ handle
+--   corresponding 'QueryFailure's.
+--
+handler :: forall h a b. (KRPC a b, Monad h)
+        => (SockAddr -> a -> h b) -> Handler h
+handler body = (name, wrapper)
+  where
+    Method name = method :: Method a b
+    wrapper addr args =
+      case fromBEncode args of
+        Left  e -> return $ Left e
+        Right a -> do
+          r <- body addr a
+          return $ Right $ toBEncode r
+
+runHandler :: MonadKRPC h m
+           => HandlerBody h -> SockAddr -> KQuery -> m KResult
+runHandler h addr KQuery {..} = Lifted.catches wrapper failbacks
+  where
+    signature = querySignature queryMethod queryId addr
+
+    wrapper = do
+      $(logDebugS) "handler.quered" signature
+      result <- liftHandler (h addr queryArgs)
+
+      case result of
+        Left msg -> do
+          $(logDebugS) "handler.bad_query" $ signature <> " !" <> T.pack msg
+          return $ Left $ KError ProtocolError (BC.pack msg) queryId
+
+        Right a -> do
+          $(logDebugS) "handler.success" signature
+          return $ Right $ KResponse a queryId (Just $ ReflectedIP addr)
+
+    failbacks =
+      [ E.Handler $ \ (e :: HandlerFailure) -> do
+          $(logDebugS) "handler.failed" signature
+          return $ Left $ KError ProtocolError (prettyHF e) queryId
+
+        -- may happen if handler makes query and fail
+      , E.Handler $ \ (e :: QueryFailure) -> do
+          return $ Left $ KError ServerError (prettyQF e) queryId
+
+        -- since handler thread exit after sendMessage we can safely
+        -- suppress async exception here
+      , E.Handler $ \ (e :: SomeException) -> do
+          return $ Left $ KError GenericError (BC.pack (show e)) queryId
+      ]
+
+dispatchHandler :: MonadKRPC h m => KQuery -> SockAddr -> m KResult
+dispatchHandler q @ KQuery {..} addr = do
+  Manager {..} <- getManager
+  case L.lookup queryMethod handlers of
+    Nothing -> return $ Left $ KError MethodUnknown queryMethod queryId
+    Just h  -> runHandler h addr q
+
+{-----------------------------------------------------------------------
+--  Listener
+-----------------------------------------------------------------------}
+
+-- TODO bound amount of parallel handler *threads*:
+--
+-- peer A flooding with find_node
+-- peer B trying to ping peer C
+-- peer B fork too many threads
+-- ... space leak
+--
+handleQuery :: MonadKRPC h m => BValue -> KQuery -> SockAddr -> m ()
+handleQuery raw q addr = void $ fork $ do
+  Manager {..} <- getManager
+  res <- dispatchHandler q addr
+  let resbe = either toBEncode toBEncode res
+  $(logOther "q") $ T.unlines
+    [ either (const "<unicode-fail>") id $ T.decodeUtf8' (BL.toStrict $ showBEncode raw)
+    , "==>"
+    , either (const "<unicode-fail>") id $ T.decodeUtf8' (BL.toStrict $ showBEncode resbe)
+    ]
+  sendMessage sock addr resbe
+
+handleResponse :: MonadKRPC h m => BValue -> KResult -> SockAddr -> m ()
+handleResponse raw result addr = do
+  Manager {..} <- getManager
+  liftIO $ do
+    let resultId = either errorId respId  result
+    mcall <- unregisterQuery (resultId, addr) pendingCalls
+    case mcall of
+      Nothing   -> return ()
+      Just ares -> putMVar ares (raw,result)
+
+handleMessage :: MonadKRPC h m => BValue -> KMessage -> SockAddr -> m ()
+handleMessage raw (Q q) = handleQuery    raw q
+handleMessage raw (R r) = handleResponse raw (Right r)
+handleMessage raw (E e) = handleResponse raw (Left  e)
+
+listener :: MonadKRPC h m => m ()
+listener = do
+  Manager {..} <- getManager
+  forever $ do
+    (bs, addr) <- liftIO $ do
+      handle exceptions $ BS.recvFrom sock (optMaxMsgSize options)
+
+    case BE.parse bs >>= \r -> (,) r <$> BE.decode bs of
+      -- TODO ignore unknown messages at all?
+      Left  e       -> liftIO $ sendMessage sock addr $ unknownMessage e
+      Right (raw,m) -> handleMessage raw m addr
+  where
+    exceptions :: IOError -> IO (BS.ByteString, SockAddr)
+    exceptions e
+        -- packets with empty payload may trigger eof exception
+      | isEOFError e = return  ("", SockAddrInet 0 0)
+      |   otherwise  = throwIO e
+
+-- | Should be run before any 'query', otherwise they will never
+-- succeed.
+listen :: MonadKRPC h m => m ()
+listen = do
+  Manager {..} <- getManager
+  tid <- fork $ do
+    listener `Lifted.finally`
+      liftIO (takeMVar listenerThread)
+  liftIO $ putMVar listenerThread tid
diff --git a/src/Network/KRPC/Message.hs b/src/Network/KRPC/Message.hs
new file mode 100644
index 00000000..6f4ae620
--- /dev/null
+++ b/src/Network/KRPC/Message.hs
@@ -0,0 +1,289 @@
+-- |
+--   Copyright   :  (c) Sam Truzjan 2013, 2014
+--   License     :  BSD3
+--   Maintainer  :  pxqr.sta@gmail.com
+--   Stability   :  experimental
+--   Portability :  portable
+--
+--   KRPC messages types used in communication. All messages are
+--   encoded as bencode dictionary.
+--
+--   Normally, you don't need to import this module.
+--
+--   See <http://www.bittorrent.org/beps/bep_0005.html#krpc-protocol>
+--
+{-# LANGUAGE OverloadedStrings      #-}
+{-# LANGUAGE FlexibleContexts       #-}
+{-# LANGUAGE TypeSynonymInstances   #-}
+{-# LANGUAGE MultiParamTypeClasses  #-}
+{-# LANGUAGE FunctionalDependencies #-}
+{-# LANGUAGE DefaultSignatures      #-}
+{-# LANGUAGE DeriveDataTypeable     #-}
+module Network.KRPC.Message
+       ( -- * Transaction
+         TransactionId
+
+         -- * Error
+       , ErrorCode (..)
+       , KError(..)
+       , decodeError
+       , unknownMessage
+
+         -- * Query
+       , KQuery(..)
+       , MethodName
+
+         -- * Response
+       , KResponse(..)
+       , ReflectedIP(..)
+
+         -- * Message
+       , KMessage (..)
+       ) where
+
+import Control.Applicative
+import Control.Arrow
+import Control.Exception.Lifted as Lifted
+import Data.BEncode as BE
+import Data.ByteString as B
+import Data.ByteString.Char8 as BC
+import qualified Data.Serialize as S
+import Data.Word
+import Data.Typeable
+import Network.Socket (SockAddr (..),PortNumber,HostAddress)
+
+
+-- | This transaction ID is generated by the querying node and is
+-- echoed in the response, so responses may be correlated with
+-- multiple queries to the same node. The transaction ID should be
+-- encoded as a short string of binary numbers, typically 2 characters
+-- are enough as they cover 2^16 outstanding queries.
+type TransactionId = ByteString
+
+unknownTransaction :: TransactionId
+unknownTransaction = ""
+
+{-----------------------------------------------------------------------
+-- Error messages
+-----------------------------------------------------------------------}
+
+-- | Types of RPC errors.
+data ErrorCode
+    -- | Some error doesn't fit in any other category.
+  = GenericError
+
+    -- | Occur when server fail to process procedure call.
+  | ServerError
+
+    -- | Malformed packet, invalid arguments or bad token.
+  | ProtocolError
+
+    -- | Occur when client trying to call method server don't know.
+  | MethodUnknown
+    deriving (Show, Read, Eq, Ord, Bounded, Typeable)
+
+-- | According to the table:
+-- <http://bittorrent.org/beps/bep_0005.html#errors>
+instance Enum ErrorCode where
+  fromEnum GenericError  = 201
+  fromEnum ServerError   = 202
+  fromEnum ProtocolError = 203
+  fromEnum MethodUnknown = 204
+  {-# INLINE fromEnum #-}
+
+  toEnum 201 = GenericError
+  toEnum 202 = ServerError
+  toEnum 203 = ProtocolError
+  toEnum 204 = MethodUnknown
+  toEnum _   = GenericError
+  {-# INLINE toEnum #-}
+
+instance BEncode ErrorCode where
+  toBEncode = toBEncode . fromEnum
+  {-# INLINE toBEncode #-}
+
+  fromBEncode b = toEnum <$> fromBEncode b
+  {-# INLINE fromBEncode #-}
+
+-- | Errors are sent when a query cannot be fulfilled. Error message
+--   can be send only from server to client but not in the opposite
+--   direction.
+--
+data KError = KError
+  { errorCode    :: !ErrorCode     -- ^ the type of error;
+  , errorMessage :: !ByteString    -- ^ human-readable text message;
+  , errorId      :: !TransactionId -- ^ match to the corresponding 'queryId'.
+  } deriving (Show, Read, Eq, Ord, Typeable)
+
+-- | Errors, or KRPC message dictionaries with a \"y\" value of \"e\",
+--   contain one additional key \"e\". The value of \"e\" is a
+--   list. The first element is an integer representing the error
+--   code. The second element is a string containing the error
+--   message.
+--
+--   Example Error Packet:
+--
+--   > { "t": "aa", "y":"e", "e":[201, "A Generic Error Ocurred"]}
+--
+--   or bencoded:
+--
+--   > d1:eli201e23:A Generic Error Ocurrede1:t2:aa1:y1:ee
+--
+instance BEncode KError where
+  toBEncode KError {..} = toDict $
+       "e" .=! (errorCode, errorMessage)
+    .: "t" .=! errorId
+    .: "y" .=! ("e" :: ByteString)
+    .: endDict
+  {-# INLINE toBEncode #-}
+
+  fromBEncode = fromDict $ do
+    lookAhead $ match "y" (BString "e")
+    (code, msg) <- field (req "e")
+    KError code msg <$>! "t"
+  {-# INLINE fromBEncode #-}
+
+instance Exception KError
+
+-- | Received 'queryArgs' or 'respVals' can not be decoded.
+decodeError :: String -> TransactionId -> KError
+decodeError msg = KError ProtocolError (BC.pack msg)
+
+-- | A remote node has send some 'KMessage' this node is unable to
+-- decode.
+unknownMessage :: String -> KError
+unknownMessage msg = KError ProtocolError (BC.pack msg) unknownTransaction
+
+{-----------------------------------------------------------------------
+-- Query messages
+-----------------------------------------------------------------------}
+
+type MethodName = ByteString
+
+-- | Query used to signal that caller want to make procedure call to
+-- callee and pass arguments in. Therefore query may be only sent from
+-- client to server but not in the opposite direction.
+--
+data KQuery = KQuery
+  { queryArgs   :: !BValue         -- ^ values to be passed to method;
+  , queryMethod :: !MethodName     -- ^ method to call;
+  , queryId     :: !TransactionId  -- ^ one-time query token.
+  } deriving (Show, Read, Eq, Ord, Typeable)
+
+-- | Queries, or KRPC message dictionaries with a \"y\" value of
+-- \"q\", contain two additional keys; \"q\" and \"a\". Key \"q\" has
+-- a string value containing the method name of the query. Key \"a\"
+-- has a dictionary value containing named arguments to the query.
+--
+--    Example Query packet:
+--
+--    > { "t" : "aa", "y" : "q", "q" : "ping", "a" : { "msg" : "hi!" } }
+--
+instance BEncode KQuery where
+  toBEncode KQuery {..} = toDict $
+       "a" .=! queryArgs
+    .: "q" .=! queryMethod
+    .: "t" .=! queryId
+    .: "y" .=! ("q" :: ByteString)
+    .: endDict
+  {-# INLINE toBEncode #-}
+
+  fromBEncode = fromDict $ do
+    lookAhead $ match "y" (BString "q")
+    KQuery <$>! "a" <*>! "q" <*>! "t"
+  {-# INLINE fromBEncode #-}
+
+newtype ReflectedIP = ReflectedIP SockAddr
+  deriving (Eq, Ord, Show)
+
+instance BEncode ReflectedIP where
+  toBEncode (ReflectedIP addr) = BString (encodeAddr addr)
+  fromBEncode (BString bs) = ReflectedIP <$> decodeAddr bs
+  fromBEncode _            = Left "ReflectedIP should be a bencoded string"
+
+port16 :: Word16 -> PortNumber
+port16 = fromIntegral
+
+decodeAddr :: ByteString -> Either String SockAddr
+decodeAddr bs | B.length bs == 6
+    = ( \(a,p) -> SockAddrInet <$> fmap port16 p <*> a )
+      $ (S.runGet S.getWord32host *** S.decode )
+      $ B.splitAt 4 bs
+decodeAddr bs | B.length bs == 18
+    = ( \(a,p) -> flip SockAddrInet6 0 <$> fmap port16 p <*> a <*> pure 0 )
+      $ (S.decode *** S.decode )
+      $ B.splitAt 16 bs
+decodeAddr _ = Left "incorrectly sized address and port"
+
+encodeAddr :: SockAddr -> ByteString
+encodeAddr (SockAddrInet port addr)
+    = S.runPut (S.putWord32host addr >> S.put (fromIntegral port :: Word16))
+encodeAddr (SockAddrInet6 port _ addr _)
+    = S.runPut (S.put addr >> S.put (fromIntegral port :: Word16))
+encodeAddr _ = B.empty
+
+{-----------------------------------------------------------------------
+-- Response messages
+-----------------------------------------------------------------------}
+
+-- | Response messages are sent upon successful completion of a
+-- query:
+--
+--   * KResponse used to signal that callee successufully process a
+--   procedure call and to return values from procedure.
+--
+--   * KResponse should not be sent if error occurred during RPC,
+--   'KError' should be sent instead.
+--
+--   * KResponse can be only sent from server to client.
+--
+data KResponse = KResponse
+  { respVals :: BValue         -- ^ 'BDict' containing return values;
+  , respId   :: TransactionId  -- ^ match to the corresponding 'queryId'.
+  , respIP   :: Maybe ReflectedIP
+  } deriving (Show, Eq, Ord, Typeable)
+
+-- | Responses, or KRPC message dictionaries with a \"y\" value of
+-- \"r\", contain one additional key \"r\". The value of \"r\" is a
+-- dictionary containing named return values.
+--
+--   Example Response packet:
+--
+--   > { "t" : "aa", "y" : "r", "r" : { "msg" : "you've sent: hi!" } }
+--
+instance BEncode KResponse where
+  toBEncode KResponse {..} = toDict $
+       "ip" .=? respIP
+    .: "r" .=! respVals
+    .: "t" .=! respId
+    .: "y" .=! ("r" :: ByteString)
+    .: endDict
+  {-# INLINE toBEncode #-}
+
+  fromBEncode = fromDict $ do
+    lookAhead $ match "y" (BString "r")
+    addr <- optional (field (req "ip"))
+    (\r t -> KResponse r t addr) <$>! "r" <*>! "t"
+  {-# INLINE fromBEncode #-}
+
+{-----------------------------------------------------------------------
+-- Summed messages
+-----------------------------------------------------------------------}
+
+-- | Generic KRPC message.
+data KMessage
+  = Q KQuery
+  | R KResponse
+  | E KError
+    deriving (Show, Eq)
+
+instance BEncode KMessage where
+  toBEncode (Q q) = toBEncode q
+  toBEncode (R r) = toBEncode r
+  toBEncode (E e) = toBEncode e
+
+  fromBEncode b =
+        Q <$> fromBEncode b
+    <|> R <$> fromBEncode b
+    <|> E <$> fromBEncode b
+    <|> decodingError "KMessage: unknown message or message tag"
diff --git a/src/Network/KRPC/Method.hs b/src/Network/KRPC/Method.hs
new file mode 100644
index 00000000..916b38a8
--- /dev/null
+++ b/src/Network/KRPC/Method.hs
@@ -0,0 +1,87 @@
+-- |
+--   Copyright   :  (c) Sam Truzjan 2013, 2014
+--   License     :  BSD3
+--   Maintainer  :  pxqr.sta@gmail.com
+--   Stability   :  experimental
+--   Portability :  portable
+--
+--   Normally, you don't need to import this module.
+--
+{-# LANGUAGE RankNTypes                 #-}
+{-# LANGUAGE MultiParamTypeClasses      #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE ScopedTypeVariables        #-}
+{-# LANGUAGE DefaultSignatures          #-}
+module Network.KRPC.Method
+       ( Method (..)
+       , KRPC (..)
+       ) where
+
+import Data.BEncode (BEncode)
+import Data.ByteString.Char8 as BC
+import Data.Char
+import Data.Monoid
+import Data.List as L
+import Data.String
+import Data.Typeable
+import Network.KRPC.Message
+
+
+-- | Method datatype used to describe method name, parameters and
+-- return values of procedure. Client use a method to /invoke/, server
+-- /implements/ the method to make the actual work.
+--
+--   We use the following fantom types to ensure type-safiety:
+--
+--     * param: Type of method parameters.
+--
+--     * result: Type of return value of the method.
+--
+newtype Method param result = Method { methodName :: MethodName }
+  deriving (Eq, Ord, IsString, BEncode)
+
+-- | Example:
+--
+--   @show (Method \"concat\" :: [Int] Int) == \"concat :: [Int] -> Int\"@
+--
+instance (Typeable a, Typeable b) => Show (Method a b) where
+  showsPrec _ = showsMethod
+
+showsMethod :: forall a b. ( Typeable a , Typeable b ) => Method a b -> ShowS
+showsMethod (Method name) =
+    showString (BC.unpack name) <>
+    showString " :: " <>
+    shows paramsTy <>
+    showString " -> " <>
+    shows valuesTy
+  where
+    impossible = error "KRPC.showsMethod: impossible"
+    paramsTy = typeOf (impossible :: a)
+    valuesTy = typeOf (impossible :: b)
+
+-- | In order to perform or handle KRPC query you need to provide
+--   corresponding 'KRPC' class.
+--
+--   Example:
+--
+--   @
+--   data Ping = Ping Text deriving BEncode
+--   data Pong = Pong Text deriving BEncode
+--
+--   instance 'KRPC' Ping Pong where
+--     method = \"ping\"
+--   @
+--
+class (Typeable req, BEncode req, Typeable resp, BEncode resp)
+    => KRPC req resp where
+
+  -- | Method name. Default implementation uses lowercased @req@
+  -- datatype name.
+  --
+  method :: Method req resp
+
+  -- TODO add underscores
+  default method :: Typeable req => Method req resp
+  method = Method $ fromString $ L.map toLower $ show $ typeOf hole
+    where
+      hole = error "krpc.method: impossible" :: req