3 files changed, 211 insertions, 1 deletions

diff --git a/src/Network/BitTorrent/Exchange/Bus.hs b/src/Network/BitTorrent/Exchange/Bus.hs
new file mode 100644
index 00000000..4800c4a0
--- /dev/null
+++ b/src/Network/BitTorrent/Exchange/Bus.hs
@@ -0,0 +1,63 @@
+module Network.BitTorrent.Exchange.Bus ( ) where
+
+type PeerWire = ConduitM Message Message IO
+
+runPeerWire :: Socket -> PeerWire () -> IO ()
+runPeerWire sock action =
+  sourceSocket sock     $=
+    S.conduitGet S.get  $=
+--    B.conduitDecode     $=
+      action            $=
+    S.conduitPut S.put  $$
+--    B.conduitEncode     $$
+  sinkSocket sock
+
+awaitMessage :: P2P Message
+awaitMessage = P2P $ ReaderT $ const $ {-# SCC awaitMessage #-} go
+  where
+    go = await >>= maybe (monadThrow PeerDisconnected) return
+{-# INLINE awaitMessage #-}
+
+yieldMessage :: Message -> P2P ()
+yieldMessage msg = P2P $ ReaderT $ const $ {-# SCC yieldMessage #-} do
+  C.yield msg
+{-# INLINE yieldMessage #-}
+
+-- TODO send vectored
+flushPending :: P2P ()
+flushPending = {-# SCC flushPending #-} do
+  session <- ask
+  queue   <- liftIO (getPending session)
+  mapM_ yieldMessage queue
+
+{-----------------------------------------------------------------------
+    P2P monad
+-----------------------------------------------------------------------}
+
+-- |
+--   Exceptions:
+--
+--     * SessionException: is visible only within one peer
+--     session. Use this exception to terminate P2P session, but not
+--     the swarm session.
+--
+newtype P2P a = P2P {
+    unP2P :: ReaderT PeerSession PeerWire a
+  } deriving ( Functor, Applicative, Monad
+             , MonadIO, MonadThrow, MonadActive
+             , MonadReader PeerSession
+             )
+
+instance MonadState SessionState P2P where
+  get    = asks sessionState >>= liftIO . readIORef
+  {-# INLINE get #-}
+  put !s = asks sessionState >>= \ref -> liftIO $ writeIORef ref s
+  {-# INLINE put #-}
+
+runP2P :: (Socket, PeerSession) -> P2P () -> IO ()
+runP2P (sock, ses) action =
+  handle isIOException $
+    runPeerWire sock (runReaderT (unP2P action) ses)
+  where
+    isIOException :: IOException -> IO ()
+    isIOException _ = return ()
diff --git a/src/Network/BitTorrent/Exchange/Message.hs b/src/Network/BitTorrent/Exchange/Message.hs
index 4ef7baf3..4d4a97e2 100644
--- a/src/Network/BitTorrent/Exchange/Message.hs
+++ b/src/Network/BitTorrent/Exchange/Message.hs
@@ -27,7 +27,7 @@
 --
 {-# LANGUAGE TemplateHaskell   #-}
 {-# OPTIONS  -fno-warn-orphans #-}
-module Network.BitTorrent.Exchange.Protocol
+module Network.BitTorrent.Exchange.Message
        ( -- * Initial handshake
          Handshake(..)
        , handshake
diff --git a/src/Network/BitTorrent/Exchange/Session.hs b/src/Network/BitTorrent/Exchange/Session.hs
new file mode 100644
index 00000000..ffc7816e
--- /dev/null
+++ b/src/Network/BitTorrent/Exchange/Session.hs
@@ -0,0 +1,147 @@
+{-# LANGUAGE TemplateHaskell    #-}
+{-# LANGUAGE DeriveDataTypeable #-}
+module Network.BitTorrent.Exchange.Session
+       (
+       ) where
+
+import Control.Concurrent.STM
+import Control.Exception
+import Control.Lens
+import Data.Function
+import Data.IORef
+import Data.Ord
+import Data.Typeable
+import Text.PrettyPrint
+
+import Data.Torrent.Bitfield
+import Data.Torrent.InfoHash
+import Network.BitTorrent.Core.PeerAddr
+import Network.BitTorrent.Exchange.Message
+import Network.BitTorrent.Exchange.Status
+
+
+type Extension = ()
+
+-- | Peer session contain all data necessary for peer to peer
+-- communication.
+data ExchangeSession = ExchangeSession
+  { -- | Used as unique identifier of the session.
+    connectedPeerAddr :: !PeerAddr
+
+    -- | Extensions such that both peer and client support.
+  , enabledExtensions :: [Extension]
+
+    -- | Broadcast messages waiting to be sent to peer.
+  , pendingMessages    :: !(TChan   Message)
+
+    -- | Dymanic P2P data.
+  , sessionState       :: !(IORef  SessionState)
+  }
+
+instance Eq ExchangeSession where
+  (==) = (==) `on` connectedPeerAddr
+  {-# INLINE (==) #-}
+
+instance Ord ExchangeSession where
+  compare = comparing connectedPeerAddr
+  {-# INLINE compare #-}
+
+enqueueBroadcast :: ExchangeSession -> Message -> IO ()
+enqueueBroadcast = undefined
+
+dequeueBroadcast :: ExchangeSession -> IO Message
+dequeueBroadcast = undefined
+
+{-----------------------------------------------------------------------
+--  Session state
+-----------------------------------------------------------------------}
+
+data SessionState = SessionState
+  { _bitfield :: !Bitfield      -- ^ Other peer Have bitfield.
+  , _status   :: !SessionStatus -- ^ Status of both peers.
+  } deriving (Show, Eq)
+
+$(makeLenses ''SessionState)
+
+--initialSessionState :: PieceCount -> SessionState
+--initialSessionState pc = SessionState (haveNone pc) def
+
+--getSessionState :: PeerSession -> IO SessionState
+--getSessionState PeerSession {..} = readIORef sessionState
+
+{-----------------------------------------------------------------------
+--  Exceptions
+-----------------------------------------------------------------------}
+
+-- | Exceptions used to interrupt the current P2P session. This
+-- exceptions will NOT affect other P2P sessions, DHT, peer <->
+-- tracker, or any other session.
+--
+data ExchangeFailure
+  = PeerDisconnected
+  | ProtocolError  Doc
+  | UnknownTorrent InfoHash
+    deriving (Show, Typeable)
+
+instance Exception ExchangeFailure
+
+-- | Do nothing with exception, used with 'handle' or 'try'.
+isSessionException :: Monad m => ExchangeFailure -> m ()
+isSessionException _ = return ()
+
+-- | The same as 'isSessionException' but output to stdout the catched
+-- exception, for debugging purposes only.
+putSessionException :: ExchangeFailure -> IO ()
+putSessionException = print
+{-
+{-----------------------------------------------------------------------
+--  Broadcasting: Have, Cancel, Bitfield, SuggestPiece
+-----------------------------------------------------------------------}
+{-
+Here we should enqueue broadcast messages and keep in mind that:
+  * We should enqueue broadcast events as they are appear.
+  * We should yield broadcast messages as fast as we get them.
+
+these 2 phases might differ in time significantly
+
+**TODO**: do this; but only when it'll be clean which other broadcast
+messages & events we should send.
+
+1. Update client have bitfield --\____ in one transaction;
+2. Update downloaded stats     --/
+3. Signal to the all other peer about this.
+-}
+
+available :: Bitfield -> SwarmSession -> STM ()
+available bf SwarmSession {..} = {-# SCC available #-} do
+    updateProgress >> broadcast
+  where
+    updateProgress = do
+      let piLen = ciPieceLength $ tInfo $ torrentMeta
+      let bytes = piLen * BF.haveCount bf
+      modifyTVar' (currentProgress clientSession) (downloadedProgress bytes)
+
+    broadcast = mapM_ (writeTChan broadcastMessages . Have) (BF.toList bf)
+
+-- TODO compute size of messages: if it's faster to send Bitfield
+-- instead many Have do that
+
+-- Also if there is single Have message in queue then the
+-- corresponding piece is likely still in memory or disc cache,
+-- when we can send SuggestPiece.
+
+readAvail :: TChan a -> STM [a]
+readAvail chan = do
+  m <- tryReadTChan chan
+  case m of
+    Just a  -> (:) <$> pure a <*> readAvail chan
+    Nothing -> return []
+
+-- | Get pending messages queue appeared in result of asynchronously
+--   changed client state. Resulting queue should be sent to a peer
+-- immediately.
+--
+getPending :: PeerSession -> IO [Message]
+getPending PeerSession {..} = {-# SCC getPending #-} do
+  atomically (readAvail pendingMessages)
+-}
\ No newline at end of file