summaryrefslogtreecommitdiffstats
blob: 3080869c2d44aeef1cf280d7a050eeaf8d071d1e (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
108
109
110
111
112
113
114
115
116
117
118
119
#include "Swiften/LinkLocal/BonjourQuerier.h"

#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <fcntl.h>

#include "Swiften/LinkLocal/BonjourBrowseQuery.h"
#include "Swiften/LinkLocal/BonjourPublishQuery.h"
#include "Swiften/Base/foreach.h"

namespace Swift {

BonjourQuerier::BonjourQuerier() : stopRequested(false), thread(0) {
	int fds[2];
	int result = pipe(fds);
	assert(result == 0);
	interruptSelectReadSocket = fds[0];
	fcntl(interruptSelectReadSocket, F_SETFL, fcntl(interruptSelectReadSocket, F_GETFL)|O_NONBLOCK);
	interruptSelectWriteSocket = fds[1];
}

BonjourQuerier::~BonjourQuerier() {
	stop();
}

boost::shared_ptr<DNSSDBrowseQuery> BonjourQuerier::createBrowseQuery() {
	return boost::shared_ptr<DNSSDBrowseQuery>(new BonjourBrowseQuery(shared_from_this()));
}

boost::shared_ptr<DNSSDPublishQuery> BonjourQuerier::createPublishQuery(const String& name, int port, const LinkLocalServiceInfo& info) {
	return boost::shared_ptr<DNSSDPublishQuery>(new BonjourPublishQuery(name, port, info, shared_from_this()));
}

void BonjourQuerier::addRunningQuery(boost::shared_ptr<BonjourQuery> query) {
	{
		boost::lock_guard<boost::mutex> lock(runningQueriesMutex);
		runningQueries.push_back(query);
	}
	runningQueriesAvailableEvent.notify_one();
	interruptSelect();
}

void BonjourQuerier::removeRunningQuery(boost::shared_ptr<BonjourQuery> query) {
	{
		boost::lock_guard<boost::mutex> lock(runningQueriesMutex);
		runningQueries.erase(std::remove(
			runningQueries.begin(), runningQueries.end(), query), runningQueries.end());
	}
}

void BonjourQuerier::interruptSelect() {
	char c = 0;
	write(interruptSelectWriteSocket, &c, 1);
}

void BonjourQuerier::start() {
	stop();
	thread = new boost::thread(boost::bind(&BonjourQuerier::run, shared_from_this()));
}

void BonjourQuerier::stop() {
	if (thread) {
		stopRequested = true;
		runningQueries.clear(); // TODO: Is this the right thing to do?
		runningQueriesAvailableEvent.notify_one();
		interruptSelect();
		thread->join();
		delete thread;
		stopRequested = false;
	}
}

void BonjourQuerier::run() {
	while (!stopRequested) {
		fd_set fdSet;
		int maxSocket;
		{
			boost::unique_lock<boost::mutex> lock(runningQueriesMutex);
			if (runningQueries.empty()) {
				runningQueriesAvailableEvent.wait(lock);
				if (runningQueries.empty()) {
					continue;
				}
			}

			// Run all running queries
			FD_ZERO(&fdSet);
			maxSocket = interruptSelectReadSocket;
			FD_SET(interruptSelectReadSocket, &fdSet);

			foreach(const boost::shared_ptr<BonjourQuery>& query, runningQueries) {
				int socketID = query->getSocketID();
				maxSocket = std::max(maxSocket, socketID);
				FD_SET(socketID, &fdSet);
			}
		}

		if (select(maxSocket+1, &fdSet, NULL, NULL, 0) <= 0) {
			continue;
		}

		if (FD_ISSET(interruptSelectReadSocket, &fdSet)) {
			char dummy;
			while (read(interruptSelectReadSocket, &dummy, 1) > 0) {}
		}

		{
			boost::lock_guard<boost::mutex> lock(runningQueriesMutex);
			foreach(boost::shared_ptr<BonjourQuery> query, runningQueries) {
				if (FD_ISSET(query->getSocketID(), &fdSet)) {
					query->processResult();
				}
			}
		}
	}
}

}