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
|
/*
This file is part of GNUnet.
(C) 2009 Christian Grothoff (and other contributing authors)
GNUnet 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, or (at your
option) any later version.
GNUnet 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 GNUnet; see the file COPYING. If not, write to the
Free Software Foundation, Inc., 59 Temple Place - Suite 330,
Boston, MA 02111-1307, USA.
*/
package org.gnunet.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* A connection to a gnunet service.
*
* Wraps a Connection, and is responsible for waiting until the underlying connection has been made
* and allows reconnects.
*/
public class Client {
private static final Logger logger = LoggerFactory
.getLogger(Client.class);
/**
* Underlying connection to the service.
*/
private Connection connection;
/**
* Host this client should be connected to.
*/
private final int port;
private final String hostname;
/**
* Initial value for connectBackoff.
*
*/
private static final RelativeTime INITAL_BACKOFF = RelativeTime.MILLISECOND.multiply(5);
/**
* Maximum value for connectBackoff.
*
*/
private static final RelativeTime MAX_BACKOFF = RelativeTime.SECOND.multiply(5);
/**
* The time to wait after an error occured while connecting.
* Every time an error occurs while connecting, this value is doubled until its maximum
* value (MAX_BACKOFF) has been reached. This strategy is called exponential backoff.
*/
private RelativeTime connectBackoff = INITAL_BACKOFF;
/**
* True if we are waiting for the client to connect before we can ask it to do
* notifyTransmitReady.
*/
private boolean notifyTransmitReadyDelayed;
/**
* Create a connection to a service.
*
* @param serviceName name of the service
* @param cfg configuration to use
*/
public Client(String serviceName, Configuration cfg) {
if (cfg == null) {
throw new AssertionError("Configuration may not be null");
}
// get port of this service from the configuration
port = (int) cfg.getValueNumer(serviceName, "PORT");
// get the hostname from the configuration
hostname = cfg.getValueString(serviceName, "HOSTNAME");
if (hostname.isEmpty()) {
throw new Configuration.ConfigurationException(String.format("hostname of service '%s' empty", serviceName));
}
reconnect();
}
/**
* Create a connection to a service with the specified hostname and port.
*
* @param hostname hostname of the service
* @param port port of the service
*/
public Client(String hostname, int port) {
this.hostname = hostname;
this.port = port;
reconnect();
}
/**
* Receive one message from the service. Can only be called after sending a message to the server.
*
* @param timeout deadline after which MessageReceiver.deadline will be called
* @param receiver MessageReceiver that is responsible for the received message
*/
public Cancelable receive(RelativeTime timeout, MessageReceiver receiver) {
return connection.receive(timeout, receiver);
}
private static class DelayedTransmitHandle implements Cancelable {
Cancelable realTransmitHandle;
Cancelable timeoutHandle;
@Override
public void cancel() {
throw new UnsupportedOperationException();
}
}
/**
* Ask the client to call us once it is able to send a message.
*
*
* @param timeout after how long should we give up (and call transmitter.transmit(null))
* @param autoRetry if the connection to the service dies, should we
* automatically reconnect and retry (within the deadline period)
* or should we immediately fail in this case? Pass true
* if the caller does not care about temporary connection errors,
* for example because the protocol is stateless
* @param size size of the message we want to transmit, can be an upper bound
*@param transmitter the MessageTransmitter object to call once the client is ready to transmit or
* when the timeout is over. Guaranteed to be called *after* notifyTransmitReady has returned. @return a handle that can be used to cancel the transmit request
*/
public Cancelable notifyTransmitReady(final RelativeTime timeout,
final boolean autoRetry, int size, final MessageTransmitter transmitter) {
if (connection.isConnected()) {
return connection.notifyTransmitReady(0, timeout, transmitter);
} else {
notifyTransmitReadyDelayed = true;
final DelayedTransmitHandle delayedTransmitHandle = new DelayedTransmitHandle();
delayedTransmitHandle.timeoutHandle = Scheduler.addDelayed(connectBackoff, new Scheduler.Task() {
@Override
public void run(Scheduler.RunContext ctx) {
if (connection == null) {
return;
}
if (connection.isConnected()) {
notifyTransmitReadyDelayed = false;
connection.notifyTransmitReady(0, timeout, transmitter);
} else {
logger.debug("still not connected, retrying in {}ms", connectBackoff.getMilliseconds());
reconnect();
connectBackoff = RelativeTime.min(connectBackoff.multiply(2), MAX_BACKOFF);
Scheduler.addDelayed(connectBackoff, this);
}
}
});
return delayedTransmitHandle;
}
}
public void reconnect() {
if (connection != null) {
connection.disconnect();
}
connection = new Connection(hostname, port);
}
/**
* Disconnect from the service. Cancel all pending receive/transmit requests.
*/
public void disconnect() {
if (notifyTransmitReadyDelayed) {
logger.error("disconnecting while notifyTransmitReady is pending");
}
connection.disconnect();
connection = null;
}
public boolean isConnected() {
return (connection != null) && connection.isConnected();
}
}
|
