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
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
|
{- git-annex XMPP client
-
- Copyright 2012 Joey Hess <joey@kitenet.net>
-
- Licensed under the GNU GPL version 3 or higher.
-}
module Assistant.Threads.XMPPClient where
import Assistant.Common
import Assistant.XMPP
import Assistant.XMPP.Client
import Assistant.NetMessager
import Assistant.Types.NetMessager
import Assistant.Types.Buddies
import Assistant.XMPP.Buddies
import Assistant.Sync
import Assistant.DaemonStatus
import qualified Remote
import Utility.ThreadScheduler
import Assistant.WebApp (UrlRenderer, renderUrl)
import Assistant.WebApp.Types
import Assistant.Alert
import Assistant.Pairing
import Assistant.XMPP.Git
import Annex.UUID
import Network.Protocol.XMPP
import Control.Concurrent
import qualified Data.Text as T
import qualified Data.Set as S
import qualified Data.Map as M
import qualified Git.Branch
import Data.Time.Clock
xmppClientThread :: UrlRenderer -> NamedThread
xmppClientThread urlrenderer = NamedThread "XMPPClient" $
restartableClient . xmppClient urlrenderer =<< getAssistant id
{- Runs the client, handing restart events. -}
restartableClient :: IO () -> Assistant ()
restartableClient a = forever $ do
tid <- liftIO $ forkIO a
waitNetMessagerRestart
liftIO $ killThread tid
xmppClient :: UrlRenderer -> AssistantData -> IO ()
xmppClient urlrenderer d = do
v <- liftAssistant $ liftAnnex getXMPPCreds
case v of
Nothing -> noop -- will be restarted once creds get configured
Just c -> retry (runclient c) =<< getCurrentTime
where
liftAssistant = runAssistant d
inAssistant = liftIO . liftAssistant
{- When the client exits, it's restarted;
- if it keeps failing, back off to wait 5 minutes before
- trying it again. -}
retry client starttime = do
e <- client
now <- getCurrentTime
if diffUTCTime now starttime > 300
then do
liftAssistant $ debug ["connection lost; reconnecting", show e]
retry client now
else do
liftAssistant $ debug ["connection failed; will retry", show e]
threadDelaySeconds (Seconds 300)
retry client =<< getCurrentTime
runclient c = liftIO $ connectXMPP c $ \jid -> do
selfjid <- bindJID jid
putStanza gitAnnexSignature
inAssistant $ debug ["connected", show selfjid]
{- The buddy list starts empty each time
- the client connects, so that stale info
- is not retained. -}
void $ inAssistant $
updateBuddyList (const noBuddies) <<~ buddyList
xmppThread $ receivenotifications selfjid
forever $ do
a <- inAssistant $ relayNetMessage selfjid
a
receivenotifications selfjid = forever $ do
l <- decodeStanza selfjid <$> getStanza
-- inAssistant $ debug ["received:", show l]
mapM_ (handle selfjid) l
handle _ (PresenceMessage p) = void $ inAssistant $
updateBuddyList (updateBuddies p) <<~ buddyList
handle _ (GotNetMessage QueryPresence) = putStanza gitAnnexSignature
handle _ (GotNetMessage (NotifyPush us)) = void $ inAssistant $ pull us
handle selfjid (GotNetMessage (PairingNotification stage c u)) =
maybe noop (inAssistant . pairMsgReceived urlrenderer stage u selfjid) (parseJID c)
handle _ (GotNetMessage m@(CanPush _)) = handlepushmsg m
handle _ (GotNetMessage m@(PushRequest _)) = handlepushmsg m
handle _ (GotNetMessage m@(StartingPush _)) = handlepushmsg m
handle _ (GotNetMessage m) = void $ inAssistant $ queueNetPushMessage m
handle _ (Ignorable _) = noop
handle _ (Unknown _) = noop
handle _ (ProtocolError _) = noop
handlepushmsg m = inAssistant $
unlessM (queueNetPushMessage m) $
void $ forkIO <~> handlePushMessage m
data XMPPEvent
= GotNetMessage NetMessage
| PresenceMessage Presence
| Ignorable ReceivedStanza
| Unknown ReceivedStanza
| ProtocolError ReceivedStanza
deriving Show
{- Decodes an XMPP stanza into one or more events. -}
decodeStanza :: JID -> ReceivedStanza -> [XMPPEvent]
decodeStanza selfjid s@(ReceivedPresence p)
| presenceType p == PresenceError = [ProtocolError s]
| presenceFrom p == Nothing = [Ignorable s]
| presenceFrom p == Just selfjid = [Ignorable s]
| otherwise = maybe [PresenceMessage p] decode (getGitAnnexAttrValue p)
where
decode (attr, v, _tag)
| attr == pushAttr = impliedp $ GotNetMessage $ NotifyPush $
decodePushNotification v
| attr == queryAttr = impliedp $ GotNetMessage QueryPresence
| otherwise = [Unknown s]
{- Things sent via presence imply a presence message,
- along with their real meaning. -}
impliedp v = [PresenceMessage p, v]
decodeStanza selfjid s@(ReceivedMessage m)
| messageFrom m == Nothing = [Ignorable s]
| messageFrom m == Just selfjid = [Ignorable s]
| messageType m == MessageError = [ProtocolError s]
| otherwise = maybe [Unknown s] decode (getGitAnnexAttrValue m)
where
decode (attr, v, tag)
| attr == pairAttr = use $ decodePairingNotification v
| attr == canPushAttr = use decodeCanPush
| attr == pushRequestAttr = use decodePushRequest
| attr == startingPushAttr = use decodeStartingPush
| attr == receivePackAttr = use $ decodeReceivePackOutput tag
| attr == sendPackAttr = use $ decodeSendPackOutput tag
| attr == receivePackDoneAttr = use $ decodeReceivePackDone v
| otherwise = [Unknown s]
use v = [maybe (Unknown s) GotNetMessage (v m)]
decodeStanza _ s = [Unknown s]
{- Waits for a NetMessager message to be sent, and relays it to XMPP. -}
relayNetMessage :: JID -> Assistant (XMPP ())
relayNetMessage selfjid = convert =<< waitNetMessage
where
convert (NotifyPush us) = return $ putStanza $ pushNotification us
convert QueryPresence = return $ putStanza presenceQuery
convert (PairingNotification stage c u) = withclient c $ \tojid -> do
changeBuddyPairing tojid True
return $ putStanza $ pairingNotification stage u tojid selfjid
convert (CanPush c) = sendclient c canPush
convert (PushRequest c) = sendclient c pushRequest
convert (StartingPush c) = sendclient c startingPush
convert (ReceivePackOutput c b) = sendclient c $ receivePackOutput b
convert (SendPackOutput c b) = sendclient c $ sendPackOutput b
convert (ReceivePackDone c code) = sendclient c $ receivePackDone code
sendclient c construct = withclient c $ \tojid ->
return $ putStanza $ construct tojid selfjid
withclient c a = case parseJID c of
Nothing -> return noop
Just tojid
| tojid == selfjid -> return noop
| otherwise -> a tojid
{- Runs a XMPP action in a separate thread, using a session to allow it
- to access the same XMPP client. -}
xmppThread :: XMPP () -> XMPP ()
xmppThread a = do
s <- getSession
void $ liftIO $ forkIO $
void $ runXMPP s a
{- We only pull from one remote out of the set listed in the push
- notification, as an optimisation.
-
- Note that it might be possible (though very unlikely) for the push
- notification to take a while to be sent, and multiple pushes happen
- before it is sent, so it includes multiple remotes that were pushed
- to at different times.
-
- It could then be the case that the remote we choose had the earlier
- push sent to it, but then failed to get the later push, and so is not
- fully up-to-date. If that happens, the pushRetryThread will come along
- and retry the push, and we'll get another notification once it succeeds,
- and pull again. -}
pull :: [UUID] -> Assistant ()
pull [] = noop
pull us = do
rs <- filter matching . syncRemotes <$> getDaemonStatus
debug $ "push notification for" : map (fromUUID . Remote.uuid ) rs
pullone rs =<< liftAnnex (inRepo Git.Branch.current)
where
matching r = Remote.uuid r `S.member` s
s = S.fromList us
pullone [] _ = noop
pullone (r:rs) branch =
unlessM (all id . fst <$> manualPull branch [r]) $
pullone rs branch
pairMsgReceived :: UrlRenderer -> PairStage -> UUID -> JID -> JID -> Assistant ()
pairMsgReceived urlrenderer PairReq theiruuid selfjid theirjid
-- PairReq from another client using our JID is automatically accepted.
| baseJID selfjid == baseJID theirjid = do
selfuuid <- liftAnnex getUUID
sendNetMessage $
PairingNotification PairAck (formatJID theirjid) selfuuid
finishXMPPPairing theirjid theiruuid
-- Show an alert to let the user decide if they want to pair.
| otherwise = do
let route = FinishXMPPPairR (PairKey theiruuid $ formatJID theirjid)
url <- liftIO $ renderUrl urlrenderer route []
close <- asIO1 removeAlert
void $ addAlert $ pairRequestReceivedAlert (T.unpack $ buddyName theirjid)
AlertButton
{ buttonUrl = url
, buttonLabel = T.pack "Respond"
, buttonAction = Just close
}
pairMsgReceived _ PairAck theiruuid _selfjid theirjid =
{- PairAck must come from one of the buddies we are pairing with;
- don't pair with just anyone. -}
whenM (isBuddyPairing theirjid) $ do
changeBuddyPairing theirjid False
selfuuid <- liftAnnex getUUID
sendNetMessage $
PairingNotification PairDone (formatJID theirjid) selfuuid
finishXMPPPairing theirjid theiruuid
pairMsgReceived _ PairDone _theiruuid _selfjid theirjid =
changeBuddyPairing theirjid False
isBuddyPairing :: JID -> Assistant Bool
isBuddyPairing jid = maybe False buddyPairing <$>
getBuddy (genBuddyKey jid) <<~ buddyList
changeBuddyPairing :: JID -> Bool -> Assistant ()
changeBuddyPairing jid ispairing =
updateBuddyList (M.adjust set key) <<~ buddyList
where
key = genBuddyKey jid
set b = b { buddyPairing = ispairing }
|