blob: 0f87106a948f28db13df7991a77ae45e2b0012bd (
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
|
/*
* Copyright (C) 2008-2014 TrinityCore <http://www.trinitycore.org/>
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Free Software Foundation; either version 2 of the License, or (at your
* option) any later version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
* more details.
*
* You should have received a copy of the GNU General Public License along
* with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "ZmqMux.h"
#include "ZmqContext.h"
ZmqMux::ZmqMux(std::string const& fromUri, std::string const& toUri):
_fromAddress(fromUri)
{
printf("Opening muxer thread from %s to %s\n", fromUri.c_str(), toUri.c_str());
_from = sIpcContext->CreateNewSocket(zmqpp::socket_type::pull);
_to = sIpcContext->CreateNewSocket(zmqpp::socket_type::push);
_from->bind(fromUri);
_to->connect(toUri);
}
ZmqMux::~ZmqMux()
{
delete _from;
delete _to;
}
void ZmqMux::HandleOpen()
{
_poller->add(*_from);
_poller->add(*_to, zmqpp::poller::poll_out);
}
bool ZmqMux::Send(zmqpp::message* m, bool dont_block)
{
if (_socket.get() == nullptr)
{
_socket.reset(sIpcContext->CreateNewSocket(zmqpp::socket_type::push));
_socket->connect(_fromAddress);
}
return _socket->send(*m, dont_block);
}
void ZmqMux::Run()
{
for (;;)
{
std::this_thread::sleep_for(std::chrono::milliseconds(50));
if (!_poller->poll())
break;
if (ProcessExit())
break;
Pipeline(_from, _to);
}
}
|
