1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22 package net.grinder.communication;
23
24 import static org.junit.Assert.assertEquals;
25 import static org.junit.Assert.assertNotNull;
26 import static org.junit.Assert.assertNull;
27 import static org.junit.Assert.assertTrue;
28 import static org.junit.Assert.fail;
29
30 import java.io.InputStream;
31 import java.io.ObjectInputStream;
32 import java.io.ObjectOutputStream;
33 import java.io.OutputStream;
34 import java.net.Socket;
35
36 import net.grinder.communication.BlockingSender.NoResponseException;
37 import net.grinder.util.StreamCopier;
38
39 import org.junit.Test;
40
41
42
43
44
45
46
47 public class TestClientSender {
48
49 @Test public void testSend() throws Exception {
50
51 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
52
53 final Connector connector =
54 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
55 ConnectionType.AGENT);
56
57 final Sender clientSender = ClientSender.connect(connector, null);
58
59 socketAcceptor.join();
60
61 final SimpleMessage message1 = new SimpleMessage();
62 final SimpleMessage message2 = new SimpleMessage();
63
64 clientSender.send(message1);
65 clientSender.send(message2);
66
67 final InputStream socketInput =
68 socketAcceptor.getAcceptedSocket().getInputStream();
69
70 assertEquals(ConnectionType.AGENT,
71 Connector.read(socketInput).getConnectionType());
72
73
74
75 final ObjectInputStream inputStream1 = new ObjectInputStream(socketInput);
76 final Object o1 = inputStream1.readObject();
77
78 final ObjectInputStream inputStream2 = new ObjectInputStream(socketInput);
79 final Object o2 = inputStream2.readObject();
80
81 assertEquals(message1, o1);
82 assertEquals(message2, o2);
83
84 assertEquals(0, socketInput.available());
85
86 socketAcceptor.close();
87
88 try {
89 ClientReceiver.connect(connector, new StubAddress());
90 fail("Expected CommunicationException");
91 }
92 catch (CommunicationException e) {
93 }
94 }
95
96 @Test public void testShutdown() throws Exception {
97
98 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
99
100 final Connector connector =
101 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
102 ConnectionType.AGENT);
103
104 final Sender clientSender = ClientSender.connect(connector, null);
105
106 socketAcceptor.join();
107
108 final Message message = new SimpleMessage();
109
110 clientSender.send(message);
111
112 clientSender.shutdown();
113
114 try {
115 clientSender.send(message);
116 fail("Expected CommunicationException");
117 }
118 catch (CommunicationException e) {
119 }
120
121 final InputStream socketInput =
122 socketAcceptor.getAcceptedSocket().getInputStream();
123
124 assertEquals(ConnectionType.AGENT,
125 Connector.read(socketInput).getConnectionType());
126
127 final ObjectInputStream inputStream1 = new ObjectInputStream(socketInput);
128 final Object o1 = inputStream1.readObject();
129 assertNotNull(o1);
130
131 final ObjectInputStream inputStream2 = new ObjectInputStream(socketInput);
132 final Object o2 = inputStream2.readObject();
133
134 assertTrue(o2 instanceof CloseCommunicationMessage);
135
136 socketAcceptor.close();
137 }
138
139 @Test public void testShutdownPeerDifferently() throws Exception {
140 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
141
142 final Connector connector =
143 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
144 ConnectionType.AGENT);
145
146 final ClientSender clientSender = ClientSender.connect(connector, null);
147
148 socketAcceptor.join();
149
150 new StreamSender(socketAcceptor.getAcceptedSocket().getOutputStream())
151 .writeMessage(new CloseCommunicationMessage());
152
153 try {
154 clientSender.blockingSend(new SimpleMessage());
155 fail("Expected CommunicationException");
156 }
157 catch (CommunicationException e) {
158 }
159
160 socketAcceptor.close();
161 }
162
163 @Test public void testWithPairedClientReceiver() throws Exception {
164 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
165
166 final Connector connector =
167 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
168 ConnectionType.AGENT);
169
170 final ClientReceiver clientReceiver =
171 ClientReceiver.connect(connector, new StubAddress());
172 final ClientSender clientSender = ClientSender.connect(clientReceiver);
173
174 socketAcceptor.join();
175
176
177 final Socket remoteSocket = socketAcceptor.getAcceptedSocket();
178 final InputStream inputStream = remoteSocket.getInputStream();
179 assertEquals(ConnectionType.AGENT,
180 Connector.read(inputStream).getConnectionType());
181
182 new Thread(
183 new StreamCopier(1000, true).getRunnable(inputStream,
184 remoteSocket.getOutputStream()),
185 "Echo stream").start();
186
187 final Message message = new SimpleMessage();
188 clientSender.send(message);
189 final Message receivedMessage = clientReceiver.waitForMessage();
190
191 assertEquals(receivedMessage, message);
192
193 clientReceiver.shutdown();
194 }
195
196 @Test public void testWithBadPairedClientReceiver() throws Exception {
197 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
198
199 final Connector connector =
200 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
201 ConnectionType.AGENT);
202
203 final ClientReceiver clientReceiver =
204 ClientReceiver.connect(connector, new StubAddress());
205 clientReceiver.shutdown();
206
207
208 final ClientSender sender = ClientSender.connect(clientReceiver);
209
210 try {
211 sender.send(null);
212 fail("Expected CommunicationException");
213 }
214 catch (CommunicationException e) {
215 }
216 }
217
218 @Test public void testBlockingSend() throws Exception {
219
220 final SocketAcceptorThread socketAcceptor = SocketAcceptorThread.create();
221
222 final Connector connector =
223 new Connector(socketAcceptor.getHostName(), socketAcceptor.getPort(),
224 ConnectionType.AGENT);
225
226 final BlockingSender clientSender = ClientSender.connect(connector, null);
227
228 socketAcceptor.join();
229
230 final Socket acceptedSocket = socketAcceptor.getAcceptedSocket();
231 final InputStream socketInput = acceptedSocket.getInputStream();
232 final OutputStream socketOutput = acceptedSocket.getOutputStream();
233
234 assertEquals(ConnectionType.AGENT,
235 Connector.read(socketInput).getConnectionType());
236
237 final SimpleMessage message1 = new SimpleMessage();
238
239 final ReceiveOneMessageAndReply receiver1 =
240 new ReceiveOneMessageAndReply(socketInput, socketOutput);
241 receiver1.start();
242
243 final Object received1 = clientSender.blockingSend(message1);
244 assertEquals(message1, received1);
245 receiver1.join();
246 assertNull(receiver1.getException());
247
248 final NoResponseMessage message2 = new NoResponseMessage();
249
250 final ReceiveOneMessageAndReply receiver2 =
251 new ReceiveOneMessageAndReply(socketInput, socketOutput);
252 receiver2.start();
253
254 try {
255 clientSender.blockingSend(message2);
256 fail("Expected NoResponseException");
257 }
258 catch (NoResponseException e) {
259 }
260
261 receiver2.join();
262 assertNull(receiver2.getException());
263
264 socketAcceptor.close();
265 }
266
267 private static final class ReceiveOneMessageAndReply extends Thread {
268
269 private final InputStream m_inputStream;
270 private final OutputStream m_outputStream;
271 private Exception m_exception;
272
273 public ReceiveOneMessageAndReply(InputStream inputStream,
274 OutputStream outputStream) {
275 m_inputStream = inputStream;
276 m_outputStream = outputStream;
277 }
278
279 public void run() {
280
281
282 try {
283 final ObjectInputStream inputStream =
284 new ObjectInputStream(m_inputStream);
285 final MessageRequiringResponse responseSender =
286 (MessageRequiringResponse)inputStream.readObject();
287
288 assert m_inputStream.available() == 0;
289
290 final ObjectOutputStream outputStream =
291 new ObjectOutputStream(m_outputStream);
292 outputStream.writeObject(responseSender.getMessage());
293 outputStream.flush();
294 }
295 catch (Exception e) {
296 m_exception = e;
297 }
298 }
299
300 public Exception getException() {
301 return m_exception;
302 }
303 }
304 }