1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
|
-- |
-- Copyright : (c) Sam T. 2013
-- License : MIT
-- Maintainer : pxqr.sta@gmail.com
-- Stability : experimental
-- Portability : portable
--
-- This module implement opaque broadcast message passing. It
-- provides sessions needed by Network.BitTorrent and
-- Network.BitTorrent.Exchange and modules. To hide some internals
-- of this module we detach it from Exchange.
--
{-# LANGUAGE RecordWildCards #-}
module Network.BitTorrent.Internal
( Progress(..), startProgress
, ClientSession(..), newClient
, SwarmSession(..), newLeacher, newSeeder
, PeerSession(..), withPeerSession
-- * Timeouts
, updateIncoming, updateOutcoming
) where
import Control.Applicative
import Control.Concurrent
import Control.Concurrent.STM
import Control.Exception
import Data.IORef
import Data.Function
import Data.Ord
import Data.Set as S
import Data.Conduit
import Data.Conduit.Cereal
import Data.Conduit.Network
import Data.Serialize
import Network
import Network.Socket
import Network.Socket.ByteString
import GHC.Event as Ev
import Data.Bitfield as BF
import Data.Torrent
import Network.BitTorrent.Extension
import Network.BitTorrent.Peer
import Network.BitTorrent.Exchange.Protocol as BT
-- | 'Progress' contains upload/download/left stats about
-- current client state.
--
-- This data is considered as dynamic within one session.
--
data Progress = Progress {
prUploaded :: Integer -- ^ Total amount of bytes uploaded.
, prDownloaded :: Integer -- ^ Total amount of bytes downloaded.
, prLeft :: Integer -- ^ Total amount of bytes left.
} deriving Show
startProgress :: Integer -> Progress
startProgress = Progress 0 0
{-----------------------------------------------------------------------
Client session
-----------------------------------------------------------------------}
-- | In one application you could have many clients.
data ClientSession = ClientSession {
clientPeerID :: PeerID -- ^
, allowedExtensions :: [Extension] -- ^
, swarmSessions :: TVar (Set SwarmSession)
, eventManager :: EventManager
, currentProgress :: IORef Progress
}
instance Eq ClientSession where
(==) = (==) `on` clientPeerID
instance Ord ClientSession where
compare = comparing clientPeerID
newClient :: [Extension] -> IO ClientSession
newClient exts = ClientSession <$> newPeerID
<*> pure exts
<*> newTVarIO S.empty
<*> Ev.new
<*> newIORef (startProgress 0)
{-----------------------------------------------------------------------
Swarm session
-----------------------------------------------------------------------}
-- | Extensions are set globally by
-- Swarm session are un
data SwarmSession = SwarmSession {
torrentMeta :: Torrent
, clientSession :: ClientSession
, clientBitfield :: IORef Bitfield
, connectedPeers :: TVar (Set PeerSession)
}
instance Eq SwarmSession where
(==) = (==) `on` (tInfoHash . torrentMeta)
instance Ord SwarmSession where
compare = comparing (tInfoHash . torrentMeta)
newSwarmSession :: Bitfield -> ClientSession -> Torrent -> IO SwarmSession
newSwarmSession bf cs @ ClientSession {..} t @ Torrent {..}
= SwarmSession <$> pure t
<*> pure cs
<*> newIORef bf
<*> newTVarIO S.empty
newSeeder :: ClientSession -> Torrent -> IO SwarmSession
newSeeder cs t @ Torrent {..}
= newSwarmSession (haveAll (pieceCount tInfo)) cs t
newLeacher :: ClientSession -> Torrent -> IO SwarmSession
newLeacher cs t @ Torrent {..}
= newSwarmSession (haveNone (pieceCount tInfo)) cs t
isLeacher :: SwarmSession -> IO Bool
isLeacher = undefined
{-----------------------------------------------------------------------
Peer session
-----------------------------------------------------------------------}
data PeerSession = PeerSession {
connectedPeerAddr :: PeerAddr
, swarmSession :: SwarmSession
, enabledExtensions :: [Extension]
-- | To dissconnect from died peers appropriately we should check
-- if a peer do not sent the KA message within given interval. If
-- yes, we should throw an exception in 'TimeoutCallback' and
-- close session between peers.
--
-- We should update timeout if we /receive/ any message within
-- timeout interval to keep connection up.
, incomingTimeout :: TimeoutKey
-- | To send KA message appropriately we should know when was last
-- time we sent a message to a peer. To do that we keep registered
-- timeout in event manager and if we do not sent any message to
-- the peer within given interval then we send KA message in
-- 'TimeoutCallback'.
--
-- We should update timeout if we /send/ any message within timeout
-- to avoid reduntant KA messages.
, outcomingTimeout :: TimeoutKey
, broadcastMessages :: Chan [Message]
, peerBitfield :: IORef Bitfield
, peerSessionStatus :: IORef SessionStatus
}
instance Eq PeerSession where
(==) = (==) `on` connectedPeerAddr
instance Ord PeerSession where
compare = comparing connectedPeerAddr
-- TODO check if it connected yet peer
withPeerSession :: SwarmSession -> PeerAddr
-> ((Socket, PeerSession) -> IO a)
-> IO a
withPeerSession ss @ SwarmSession {..} addr
= bracket openSession closeSession
where
openSession = do
let caps = encodeExts $ allowedExtensions $ clientSession
let pid = clientPeerID $ clientSession
let chs = Handshake defaultBTProtocol caps (tInfoHash torrentMeta) pid
sock <- connectToPeer addr
phs <- handshake sock chs `onException` close sock
let enabled = decodeExts (enabledCaps caps (handshakeCaps phs))
ps <- PeerSession addr ss enabled
<$> registerTimeout (eventManager clientSession)
maxIncomingTime abortSession
<*> registerTimeout (eventManager clientSession)
maxOutcomingTime (sendKA sock)
<*> newChan
<*> pure clientBitfield
<*> newIORef initSessionStatus
return (sock, ps)
closeSession = close . fst
{-----------------------------------------------------------------------
Timeouts
-----------------------------------------------------------------------}
sec :: Int
sec = 1000 * 1000
maxIncomingTime :: Int
maxIncomingTime = 120 * sec
maxOutcomingTime :: Int
maxOutcomingTime = 60 * sec
-- | Should be called after we have received any message from a peer.
updateIncoming :: PeerSession -> IO ()
updateIncoming PeerSession {..} = do
updateTimeout (eventManager (clientSession swarmSession))
incomingTimeout maxIncomingTime
-- | Should be called before we have send any message to a peer.
updateOutcoming :: PeerSession -> IO ()
updateOutcoming PeerSession {..} =
updateTimeout (eventManager (clientSession swarmSession))
outcomingTimeout maxOutcomingTime
sendKA :: Socket -> IO ()
sendKA sock = sendAll sock (encode BT.KeepAlive)
abortSession :: IO ()
abortSession = error "abortSession: not implemented"
|
