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
|
#ifndef SOCKETPROXY_H
#define SOCKETPROXY_H
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
#include "qpid/sys/Socket.h"
#include "qpid/sys/Poller.h"
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Thread.h"
#include "qpid/sys/Mutex.h"
#include "qpid/client/Connection.h"
#include "qpid/log/Statement.h"
#include <algorithm>
/**
* A simple socket proxy that forwards to another socket.
* Used between client & local broker to simulate network failures.
*/
class SocketProxy : private qpid::sys::Runnable
{
public:
/** Connect to connectPort on host, start a forwarding thread.
* Listen for connection on getPort().
*/
SocketProxy(int connectPort, const std::string host="localhost")
: closed(false), port(listener.listen())
{
client.connect(host, connectPort);
thread = qpid::sys::Thread(static_cast<qpid::sys::Runnable*>(this));
}
~SocketProxy() { close(); }
/** Simulate a network disconnect. */
void close() {
{
qpid::sys::Mutex::ScopedLock l(lock);
if (closed) return;
closed=true;
}
poller.shutdown();
thread.join();
client.close();
}
bool isClosed() const {
qpid::sys::Mutex::ScopedLock l(lock);
return closed;
}
uint16_t getPort() const { return port; }
private:
static void throwErrno(const std::string& msg) {
throw qpid::Exception(msg+":"+qpid::strError(errno));
}
static void throwIf(bool condition, const std::string& msg) {
if (condition) throw qpid::Exception(msg);
}
void run() {
std::auto_ptr<qpid::sys::Socket> server;
try {
qpid::sys::PollerHandle listenerHandle(listener);
poller.addFd(listenerHandle, qpid::sys::Poller::IN);
qpid::sys::Poller::Event event = poller.wait();
throwIf(event.type == qpid::sys::Poller::SHUTDOWN, "Closed by close()");
throwIf(!(event.type == qpid::sys::Poller::READABLE && event.handle == &listenerHandle), "Accept failed");
poller.delFd(listenerHandle);
server.reset(listener.accept(0, 0));
// Pump data between client & server sockets
qpid::sys::PollerHandle clientHandle(client);
qpid::sys::PollerHandle serverHandle(*server);
poller.addFd(clientHandle, qpid::sys::Poller::IN);
poller.addFd(serverHandle, qpid::sys::Poller::IN);
char buffer[1024];
for (;;) {
qpid::sys::Poller::Event event = poller.wait();
throwIf(event.type == qpid::sys::Poller::SHUTDOWN, "Closed by close()");
throwIf(event.type == qpid::sys::Poller::DISCONNECTED, "client/server disconnected");
if (event.handle == &serverHandle) {
client.write(buffer, server->read(buffer, sizeof(buffer)));
poller.rearmFd(serverHandle);
} else if (event.handle == &clientHandle) {
server->write(buffer, client.read(buffer, sizeof(buffer)));
poller.rearmFd(clientHandle);
} else {
throwIf(true, "No handle ready");
}
}
}
catch (const std::exception& e) {
QPID_LOG(debug, "SocketProxy::run exiting: " << e.what());
}
if (server.get()) server->close();
close();
}
mutable qpid::sys::Mutex lock;
bool closed;
qpid::sys::Poller poller;
qpid::sys::Socket client, listener;
uint16_t port;
qpid::sys::Thread thread;
};
#endif
|
