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
|
{-# LANGUAGE CPP #-}
{-# LANGUAGE GADTs #-}
module Connection.Tox where
import qualified Connection as G
;import Connection (Manager (..), Policy (..))
import Control.Concurrent.STM
import Control.Monad
import Data.Dependent.Sum
import Data.Functor.Identity
import qualified Data.Map as Map
import Connection.Tox.Threads
import Network.Tox.NodeId
import PingMachine
import Text.Read
#ifdef THREAD_DEBUG
import Control.Concurrent.Lifted.Instrument
#else
import Control.Concurrent.Lifted
import GHC.Conc (labelThread)
#endif
data Parameters = Parameters
{ -- | Various Tox transports and clients.
-- toxTransports :: Tox
-- | Thread to be forked when a connection is established.
-- TODO: this function should accept relevant parameters.
onToxSession :: IO ()
}
data Key = Key NodeId{-me-} NodeId{-them-}
deriving (Eq,Ord)
instance Show Key where show = show . showKey_
-- | A conneciton status that is tagged with a state type that is specific to
-- the status.
data Transient a where
IsDormant :: Transient ()
IsAwaitingDHTKey :: Transient ()
IsAcquiringIPAddress :: Transient ()
IsAcquiringCookie :: Transient ()
IsAwaitingHandshake :: Transient ()
IsAwaitingSessionPacket :: Transient ()
IsEstablished :: Transient ()
untag :: DSum Transient Identity -> G.Status ToxProgress
untag (IsDormant :=> _) = G.Dormant
untag (IsAwaitingDHTKey :=> _) = G.InProgress AwaitingDHTKey
untag (IsAcquiringIPAddress :=> _) = G.InProgress AcquiringIPAddress
untag (IsAcquiringCookie :=> _) = G.InProgress AcquiringCookie
untag (IsAwaitingHandshake :=> _) = G.InProgress AwaitingHandshake
untag (IsAwaitingSessionPacket :=> _) = G.InProgress AwaitingSessionPacket
untag (IsEstablished :=> _) = G.Established
data SessionState = SessionState
{ transient :: TVar (DSum Transient Identity)
, connPolicy :: TVar Policy
, connPingLogic :: PingMachine
}
sessionStatus :: SessionState -> G.Connection ToxProgress
sessionStatus st = G.Connection
{ G.connStatus = untag <$> readTVar (transient st)
, G.connPolicy = readTVar (connPolicy st)
, G.connPingLogic = connPingLogic st
}
lookupForPolicyChange :: TVar (Map.Map Key SessionState)
-> Key -> Policy -> IO (Maybe SessionState)
lookupForPolicyChange conmap k policy = atomically $ do
cons <- readTVar conmap
fmap join $ forM (Map.lookup k cons) $ \st -> do
p <- readTVar (connPolicy st)
writeTVar (connPolicy st) policy
return $ do
guard $ p /= policy
return st
-- | This function will fork threads as necessary.
setToxPolicy :: Parameters
-> TVar (Map.Map Key SessionState)
-> Key
-> Policy
-> IO ()
setToxPolicy params conmap k policy = case policy of
TryingToConnect -> do
mst <- lookupForPolicyChange conmap k policy
forM_ mst $ \st -> do
let getPolicy = readTVar $ connPolicy st
--TODO accept_thread may already be started if policy was OpenToConnect
accept_thread <- forkIO $ acceptContact getPolicy _accept_methods
persue_thread <- forkIO $ persueContact getPolicy _get_status _persue_methods
freshen_thread <- forkIO $ freshenContact getPolicy _get_status _freshen_methods
return ()
return ()
RefusingToConnect -> do -- disconnect or cancel any pending connection
mst <- lookupForPolicyChange conmap k policy
-- Since the 3 connection threads poll the current policy, they should
-- all terminate on their own.
return ()
OpenToConnect -> do -- passively accept connections if they initiate.
mst <- lookupForPolicyChange conmap k policy
forM_ mst $ \st -> do
let getPolicy = readTVar $ connPolicy st
accept_thread <- forkIO $ acceptContact getPolicy _accept_methods
return ()
showKey_ :: Key -> String
showKey_ (Key me them) = show me ++ ":" ++ show them
stringToKey_ :: String -> Maybe Key
stringToKey_ s = let (xs,ys) = break (==':') s
in if null ys then Nothing
else do me <- readMaybe xs
them <- readMaybe (drop 1 ys)
return $ Key me them
toxManager :: Parameters -> IO (Manager ToxProgress Key)
toxManager params = do
conmap <- newTVarIO Map.empty
return Manager
{ setPolicy = setToxPolicy params conmap -- k -> Policy -> IO ()
, connections = fmap sessionStatus <$> readTVar conmap -- STM (Map k (Connection status))
, stringToKey = stringToKey_ -- String -> Maybe k
, showProgress = show -- status -> String
, showKey = showKey_ -- k -> String
}
|
