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
|
-- |
--
-- This module defines three tasks intended to be run in separate threads:
--
-- * 'acceptContact'
--
-- * 'persueContact'
--
-- * 'freshenContact'
--
{-# LANGUAGE LambdaCase #-}
module Connection.Tox.Threads where
import Connection
import Connection.Tox
import Data.IP (IP)
import Network.Tox.Crypto.Transport
import Network.Tox.NodeId
-- import Network.Tox.DHT.Handlers (nodeSearch)
import Network.Socket
import Network.Kademlia.Search
import Network.Kademlia.Routing (BucketList)
import Control.Concurrent.STM
import Control.Monad
import Data.Function
import Data.Functor.Identity
import System.Timeout
type NodeSearch = Search NodeId (IP,PortNumber) () NodeInfo NodeInfo
data AcceptContactMethods = AcceptContactMethods
{ getHandshake :: STM (Handshake Identity)
, handshakeIsSuitable :: Handshake Identity -> STM Bool
, transitionToState :: Status ToxProgress -> STM ()
}
-- | Invokes an STM action on each incoming handshake.
--
-- Does not return until getPolicy yields RefusingToConnect.
acceptContact :: STM Policy -> AcceptContactMethods -> IO ()
acceptContact getPolicy AcceptContactMethods{..} = fix $ \loop -> do
join $ atomically $ do
orElse
(getPolicy >>= \case
RefusingToConnect -> return $ return () -- QUIT Dormant/Established
_ -> retry)
(do hs <- getHandshake
handshakeIsSuitable hs >>= \case
True -> do
-- Here we allocate a NetCrypto session for handling CryptoPacket.
transitionToState (InProgress AwaitingSessionPacket)
return loop
False -> return loop)
whileTryingAndNotEstablished :: STM Policy -> STM (Status t) -> ((Int -> IO ()) -> STM (IO ())) -> IO ()
whileTryingAndNotEstablished getPolicy getStatus body = fix $ \loop -> do
let retryWhileTrying k = getPolicy >>= \case
TryingToConnect -> retry
_ -> return k
ifEstablished t e = getStatus >>= \case
Established -> t
_ -> e
retryAfterTimeout interval = do
timeout interval $ atomically
$ orElse
(retryWhileTrying ())
(ifEstablished (return ()) retry)
loop
join $ atomically $ orElse
(retryWhileTrying (return ())) -- QUIT Dormant/Established
(ifEstablished retry
(body retryAfterTimeout))
data PersueContactMethods params = PersueContactMethods
{ getHandshakeParams :: STM params
, sendHandshake :: params -> IO ()
, retryInterval :: Int
}
-- | Continuously attempt to send handshake packets until a connection is
-- established.
--
-- As long as getPolicy is TryingToConnect and there is no established
-- connection, this function will continue.
persueContact :: STM Policy -> STM (Status t) -> PersueContactMethods a -> IO ()
persueContact getPolicy getStatus PersueContactMethods{..}
= whileTryingAndNotEstablished getPolicy getStatus
$ \retryAfterTimeout -> do
-- AwaitingDHTKey
-- AcquiringIPAddress
params <- getHandshakeParams
return $ do -- AcquiringCookie
-- AwaitingHandshake
-- AwaitingSessionPacket
sendHandshake params
retryAfterTimeout retryInterval
data FreshenContactMethods = FreshenContactMethods
{ dhtkeyInterval :: Int
, sockAddrInterval :: Int
, nodeSch :: NodeSearch
, getDHTKey :: STM (Maybe NodeId)
, getSockAddr :: STM (Maybe SockAddr)
, getBuckets :: STM (BucketList NodeInfo)
}
-- send my dht key
-- search for their sockaddr
-- monitor new dht key
-- monitor new sockaddr
--
-- Keep going while TryingToConnect
-- pause while Established
-- Useful:
-- toxidSearch onionTimeout
-- newSearch
-- searchLoop
-- searchCancel
-- -> (r -> STM Bool) -- ^ Invoked on each result. Return False to quit searching.
-- | Continuously search the DHT to obtain ip addresses and to send your dht
-- key to contacts.
--
-- As long as getPolicy is TryingToConnect and there is no established
-- connection, this function will continue.
freshenContact :: STM Policy -> STM (Status t) -> FreshenContactMethods -> IO ()
freshenContact getPolicy getStatus FreshenContactMethods{..}
= whileTryingAndNotEstablished getPolicy getStatus
-- retryAfterTimeout :: Int -> IO ()
$ \retryAfterTimeout ->
getDHTKey >>= \case
Nothing -> -- AwaitingDHTKey
retry
Just dk -> getSockAddr >>= return . \case
Nothing -> -- AcquiringIPAddress
do bkts <- atomically $ getBuckets
st <- search nodeSch bkts dk $
\r -> do -- TODO: store saddr, check for finish
return True
atomically $ searchIsFinished st >>= check
-- TODO: searchCancel on stop condition
retryAfterTimeout sockAddrInterval
Just a -> -- AcquiringCookie
-- AwaitingHandshake
-- AwaitingSessionPacket
do _todo_search_toxid_send_dhtkey -- 123 _todo_search_toxid_send_dhtkey :: IO a0
retryAfterTimeout dhtkeyInterval
|
