summaryrefslogtreecommitdiff
path: root/Assistant/NetMessager.hs
blob: c3bd73c57d46d11438ec84603adb872e86e8faf9 (plain)
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
{- git-annex assistant out of band network messager interface
 -
 - Copyright 2012 Joey Hess <joey@kitenet.net>
 -
 - Licensed under the GNU GPL version 3 or higher.
 -}

module Assistant.NetMessager where

import Assistant.Common
import Assistant.Types.NetMessager
import qualified Types.Remote as Remote
import qualified Git

import Control.Concurrent
import Control.Concurrent.STM
import Control.Concurrent.MSampleVar
import Control.Exception as E
import qualified Data.Set as S
import qualified Data.Text as T

sendNetMessage :: NetMessage -> Assistant ()
sendNetMessage m = 
	(atomically . flip writeTChan m) <<~ (netMessages . netMessager)

waitNetMessage :: Assistant (NetMessage)
waitNetMessage = (atomically . readTChan) <<~ (netMessages . netMessager)

notifyNetMessagerRestart :: Assistant ()
notifyNetMessagerRestart =
	flip writeSV () <<~ (netMessagerRestart . netMessager)

waitNetMessagerRestart :: Assistant ()
waitNetMessagerRestart = readSV <<~ (netMessagerRestart . netMessager)

getPushRunning :: Assistant PushRunning
getPushRunning =
	(atomically . readTMVar) <<~ (netMessagerPushRunning . netMessager)

{- Runs an action that runs either the send or receive end of a push.
 -
 - While the push is running, netMessagesPush will get messages put into it
 - relating to this push, while any messages relating to other pushes
 - go to netMessagesDeferred. Once the push finishes, those deferred
 - messages will be fed to handledeferred for processing.
 -}
runPush :: PushRunning -> (NetMessage -> Assistant ()) -> Assistant a -> Assistant a
runPush v handledeferred a = do
	nm <- getAssistant netMessager
	let pr = netMessagerPushRunning nm
	let setup = void $ atomically $ swapTMVar pr v
	let cleanup = atomically $ do
		void $ swapTMVar pr NoPushRunning
		emptytchan (netMessagesPush nm)
	r <- E.bracket_ setup cleanup <~> a
	(void . forkIO) <~> processdeferred nm
	return r
  where
	emptytchan c = maybe noop (const $ emptytchan c) =<< tryReadTChan c
	processdeferred nm = do
		s <- liftIO $ atomically $ swapTMVar (netMessagesDeferredPush nm) S.empty
		mapM_ rundeferred (S.toList s)
	rundeferred m = (void . (E.try :: (IO () -> IO (Either SomeException ()))))
		<~> handledeferred m

{- While a push is running, matching push messages are put into
 - netMessagesPush, while others go to netMessagesDeferredPush.
 - To avoid bloating memory, only messages that initiate pushes are
 - deferred.
 -
 - When no push is running, returns False.
 -}
queueNetPushMessage :: NetMessage -> Assistant Bool
queueNetPushMessage m = do
	nm <- getAssistant netMessager
	liftIO $ atomically $ do
		running <- readTMVar (netMessagerPushRunning nm)
		case running of
			NoPushRunning -> return False
			SendPushRunning runningcid -> do
				go nm m runningcid
				return True
			ReceivePushRunning runningcid -> do
				go nm m runningcid
				return True
  where
	go nm (Pushing cid stage) runningcid
		| cid == runningcid = writeTChan (netMessagesPush nm) m
		| isPushInitiation stage = defer nm
		| otherwise = noop
	go _ _ _ = noop

	defer nm = do
		s <- takeTMVar (netMessagesDeferredPush nm)
		putTMVar (netMessagesDeferredPush nm) $ S.insert m s

waitNetPushMessage :: Assistant (NetMessage)
waitNetPushMessage = (atomically . readTChan) <<~ (netMessagesPush . netMessager)

{- Remotes using the XMPP transport have urls like xmpp::user@host -}
isXMPPRemote :: Remote -> Bool
isXMPPRemote remote = Git.repoIsUrl r && "xmpp::" `isPrefixOf` Git.repoLocation r
  where
	r = Remote.repo remote

getXMPPClientID :: Remote -> ClientID
getXMPPClientID r = T.pack $ drop (length "xmpp::") (Git.repoLocation (Remote.repo r))