1   /*
2    *  Copyright 2006 Simon Raess
3    *
4    *  Licensed under the Apache License, Version 2.0 (the "License");
5    *  you may not use this file except in compliance with the License.
6    *  You may obtain a copy of the License at
7    *
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    * 
10   *  Unless required by applicable law or agreed to in writing, software
11   *  distributed under the License is distributed on an "AS IS" BASIS,
12   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   *  See the License for the specific language governing permissions and
14   *  limitations under the License.
15   */
16  package net.sf.beep4j.internal;
17  
18  import net.sf.beep4j.ChannelHandler;
19  import net.sf.beep4j.CloseChannelCallback;
20  import net.sf.beep4j.CloseChannelRequest;
21  import net.sf.beep4j.Message;
22  import net.sf.beep4j.MessageStub;
23  import net.sf.beep4j.NullReplyHandler;
24  import net.sf.beep4j.ReplyHandler;
25  
26  import org.jmock.Mock;
27  import org.jmock.MockObjectTestCase;
28  import org.jmock.core.Invocation;
29  import org.jmock.core.Stub;
30  
31  public class ChannelImplTest extends MockObjectTestCase {
32  	
33  	private static final String PROFILE = "http://www.example.org/profiles/echo";
34  	
35  	private static final int CHANNEL = 1;
36  	
37  	private Mock sessionMock;
38  	
39  	private InternalSession session;
40  
41  	private Mock callbackMock;
42  
43  	private CloseChannelCallback callback;
44  
45  	private Mock channelHandlerMock;
46  
47  	private ChannelHandler channelHandler;
48  		
49  	@Override
50  	protected void setUp() throws Exception {
51  		super.setUp();
52  		sessionMock = mock(InternalSession.class);
53  		session = (InternalSession) sessionMock.proxy();
54  		callbackMock = mock(CloseChannelCallback.class);
55  		callback = (CloseChannelCallback) callbackMock.proxy();
56  		channelHandlerMock = mock(ChannelHandler.class);
57  		channelHandler = (ChannelHandler) channelHandlerMock.proxy();
58  	}
59  	
60  	private void assertIsAlive(InternalChannel channel) {
61  		assertTrue(channel.isAlive());
62  		assertFalse(channel.isShuttingDown());
63  		assertFalse(channel.isDead());
64  	}
65  
66  	private void assertIsShuttingDown(InternalChannel channel) {
67  		assertFalse(channel.isAlive());
68  		assertTrue(channel.isShuttingDown());
69  		assertFalse(channel.isDead());
70  	}
71  
72  	private void assertIsDead(InternalChannel channel) {
73  		assertFalse(channel.isAlive());
74  		assertFalse(channel.isShuttingDown());
75  		assertTrue(channel.isDead());
76  	}
77  
78  	public void testAcceptedCloseRequest() throws Exception {
79  		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
80  		channel.initChannel(channelHandler);
81  		
82  		// define expectations
83  		// TODO setup ordering constraints
84  		sessionMock.expects(once()).method("requestChannelClose")
85  				.with(eq(1), ANYTHING)
86  				.will(new CloseAcceptingStub(1));
87  		
88  		callbackMock.expects(once()).method("closeAccepted");
89  		
90  		channelHandlerMock.expects(once()).method("channelClosed");
91  		
92  		// test
93  		channel.close(callback);
94  		assertIsDead(channel);
95  		try {
96  			channel.sendMessage(new MessageStub(), new NullReplyHandler());
97  			fail("sending messages in dead state must fail");
98  		} catch (IllegalStateException e) {
99  			// expected
100 		}
101 		try {
102 			channel.close(callback);
103 			fail("closing a dead channel must fail");
104 		} catch (IllegalStateException e) {
105 			// expected
106 		}
107 		
108 		verify();
109 	}
110 	
111 	public void testDelayedAcceptedCloseRequest() throws Exception {
112 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
113 		channel.initChannel(channelHandler);
114 		
115 		Message message = new MessageStub();
116 		
117 		// define expectations
118 		// TODO setup ordering constraints
119 		ParameterCaptureStub<ReplyHandler> capture = 
120 			new ParameterCaptureStub<ReplyHandler>(2, ReplyHandler.class, null);
121 		
122 		sessionMock.expects(once()).method("sendMessage")
123 				.with(eq(1), same(message), ANYTHING)
124 				.will(capture);
125 		
126 		sessionMock.expects(once()).method("requestChannelClose")
127 				.with(eq(1), ANYTHING)
128 				.will(new CloseAcceptingStub(1));
129 		
130 		callbackMock.expects(once()).method("closeAccepted");
131 		
132 		channelHandlerMock.expects(once()).method("channelClosed");
133 		
134 		// test
135 		channel.sendMessage(message, new NullReplyHandler());
136 		channel.close(callback);
137 		assertIsShuttingDown(channel);
138 		
139 		ReplyHandler listener = capture.getParameter();
140 		listener.receivedNUL();
141 		assertIsDead(channel);
142 	}
143 	
144 	public void testDeclinedCloseRequest() throws Exception {		
145 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
146 		channel.initChannel(channelHandler);
147 		
148 		Message message = new MessageStub();
149 		
150 		// define expectations
151 		// TODO: define ordering constraints
152 		sessionMock.expects(once()).method("requestChannelClose")
153 				.with(eq(1), ANYTHING)
154 				.will(new CloseDecliningStub(1, 550, "still working"));
155 		
156 		callbackMock.expects(once()).method("closeDeclined")
157 				.with(eq(550), eq("still working"));
158 		
159 		sessionMock.expects(once()).method("sendMessage")
160 				.with(eq(1), same(message), ANYTHING);
161 		
162 		// test
163 		channel.close(callback);
164 		assertIsAlive(channel);
165 		channel.sendMessage(message, new NullReplyHandler());
166 	}
167 	
168 	public void testDelayedDeclinedCloseRequest() throws Exception {
169 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
170 		channel.initChannel(channelHandler);
171 		
172 		Message m1 = new MessageStub();
173 		Message m2 = new MessageStub();
174 		
175 		// define expectations
176 		// TODO: define ordering constraints
177 		ParameterCaptureStub<ReplyHandler> capture = 
178 			new ParameterCaptureStub<ReplyHandler>(2, ReplyHandler.class, null);
179 		
180 		sessionMock.expects(once()).method("sendMessage")
181 				.with(eq(1), same(m1), ANYTHING)
182 				.will(capture);
183 		
184 		sessionMock.expects(once()).method("requestChannelClose")
185 				.with(eq(1), ANYTHING)
186 				.will(new CloseDecliningStub(1, 550, "still working"));
187 		
188 		callbackMock.expects(once()).method("closeDeclined")
189 				.with(eq(550), eq("still working"));
190 		
191 		sessionMock.expects(once()).method("sendMessage")
192 				.with(eq(1), same(m2), ANYTHING);
193 		
194 		// test
195 		channel.sendMessage(m1, new NullReplyHandler());
196 		channel.close(callback);
197 		assertIsShuttingDown(channel);
198 		
199 		ReplyHandler listener = capture.getParameter();
200 		listener.receivedNUL();
201 		assertIsAlive(channel);
202 		
203 		channel.sendMessage(m2, new NullReplyHandler());
204 	}
205 	
206 	public void testCloseRequestedAccepted() throws Exception {
207 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
208 		ChannelHandler handler = channel.initChannel(channelHandler);
209 		
210 		// define expectations
211 		channelHandlerMock.expects(once()).method("channelCloseRequested")
212 				.with(ANYTHING)
213 				.will(new CloseAcceptingRequest(0));
214 		
215 		channelHandlerMock.expects(once()).method("channelClosed");
216 		
217 		Mock mock = mock(CloseChannelRequest.class);
218 		mock.expects(once()).method("accept");
219 		CloseChannelRequest request = (CloseChannelRequest) mock.proxy();
220 		
221 		// test
222 		handler.channelCloseRequested(request);
223 		assertIsDead(channel);
224 	}
225 	
226 	public void testDelayedCloseRequestedAccepted() throws Exception {
227 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
228 		ChannelHandler handler = channel.initChannel(channelHandler);
229 		
230 		Message message = new MessageStub();
231 		
232 		ParameterCaptureStub<ReplyHandler> capture =
233 			new ParameterCaptureStub<ReplyHandler>(2, ReplyHandler.class, null);
234 		
235 		// define expectations
236 		sessionMock.expects(once()).method("sendMessage")
237 				.with(eq(1), same(message), ANYTHING)
238 				.will(capture);
239 		
240 		channelHandlerMock.expects(once()).method("channelCloseRequested")
241 				.with(ANYTHING)
242 				.will(new CloseAcceptingRequest(0));
243 		
244 		channelHandlerMock.expects(once()).method("channelClosed");
245 		
246 		Mock mock = mock(CloseChannelRequest.class);
247 		mock.expects(once()).method("accept");
248 		CloseChannelRequest request = (CloseChannelRequest) mock.proxy();
249 
250 		// test
251 		channel.sendMessage(message, new NullReplyHandler());
252 		handler.channelCloseRequested(request);
253 		assertIsShuttingDown(channel);
254 		
255 		ReplyHandler listener = capture.getParameter();
256 		listener.receivedNUL();
257 		assertIsDead(channel);
258 	}
259 	
260 	public void testCloseRequestedDeclined() throws Exception {
261 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
262 		ChannelHandler handler = channel.initChannel(channelHandler);
263 		
264 		// define expectations
265 		channelHandlerMock.expects(once()).method("channelCloseRequested")
266 				.with(ANYTHING)
267 				.will(new CloseRejectingRequest(0));
268 		
269 		Mock mock = mock(CloseChannelRequest.class);
270 		mock.expects(once()).method("reject");
271 		CloseChannelRequest request = (CloseChannelRequest) mock.proxy();
272 		
273 		// test
274 		handler.channelCloseRequested(request);
275 		assertIsAlive(channel);
276 	}
277 	
278 	public void testDelayedCloseRequestedDeclined() throws Exception {
279 		InternalChannel channel = new ChannelImpl(session, PROFILE, CHANNEL);
280 		ChannelHandler handler = channel.initChannel(channelHandler);
281 		
282 		Message m1 = new MessageStub();
283 		Message m2 = new MessageStub();
284 		
285 		ParameterCaptureStub<ReplyHandler> capture =
286 			new ParameterCaptureStub<ReplyHandler>(2, ReplyHandler.class, null);
287 		
288 		// define expectations
289 		sessionMock.expects(once()).method("sendMessage")
290 				.with(eq(1), same(m1), ANYTHING)
291 				.will(capture);
292 		
293 		channelHandlerMock.expects(once()).method("channelCloseRequested")
294 				.with(ANYTHING)
295 				.will(new CloseRejectingRequest(0));
296 		
297 		sessionMock.expects(once()).method("sendMessage")
298 				.with(eq(1), same(m2), ANYTHING);
299 		
300 		Mock mock = mock(CloseChannelRequest.class);
301 		mock.expects(once()).method("reject");
302 		CloseChannelRequest request = (CloseChannelRequest) mock.proxy();
303 
304 		// test
305 		channel.sendMessage(m1, new NullReplyHandler());
306 		handler.channelCloseRequested(request);
307 		assertIsShuttingDown(channel);
308 		
309 		ReplyHandler listener = capture.getParameter();
310 		listener.receivedNUL();
311 		assertIsAlive(channel);
312 		
313 		channel.sendMessage(m2, new NullReplyHandler());
314 	}
315 	
316 	private static class CloseAcceptingStub implements Stub {
317 		private final int index;
318 		private CloseAcceptingStub(int index) {
319 			this.index = index;
320 		}
321 		public StringBuffer describeTo(StringBuffer buf) {
322 			buf.append("stub[accept close request]");
323 			return buf;
324 		}
325 		public Object invoke(Invocation invocation) throws Throwable {
326 			CloseChannelCallback callback = (CloseChannelCallback) invocation.parameterValues.get(index);
327 			callback.closeAccepted();
328 			return null;
329 		}
330 	}
331 	
332 	private static class CloseAcceptingRequest implements Stub {
333 		private final int index;
334 		private CloseAcceptingRequest(int index) {
335 			this.index = index;
336 		}
337 		public StringBuffer describeTo(StringBuffer buf) {
338 			buf.append("stub[accept close request]");
339 			return buf;
340 		}
341 		public Object invoke(Invocation invocation) throws Throwable {
342 			CloseChannelRequest callback = (CloseChannelRequest) invocation.parameterValues.get(index);
343 			callback.accept();
344 			return null;
345 		}
346 	}
347 	
348 	private static class CloseDecliningStub implements Stub {
349 		private final int code;
350 		private final String message;
351 		private final int index;
352 		private CloseDecliningStub(int index, int code, String message) {
353 			this.code = code;
354 			this.message = message;
355 			this.index = index;
356 		}
357 		public StringBuffer describeTo(StringBuffer buf) {
358 			buf.append("stub[decline close request]");
359 			return buf;
360 		}
361 		public Object invoke(Invocation invocation) throws Throwable {
362 			CloseChannelCallback callback = (CloseChannelCallback) invocation.parameterValues.get(index);
363 			callback.closeDeclined(code, message);
364 			return null;
365 		}
366 	}
367 	
368 	private static class CloseRejectingRequest implements Stub {
369 		private final int index;
370 		private CloseRejectingRequest(int index) {
371 			this.index = index;
372 		}
373 		public StringBuffer describeTo(StringBuffer buf) {
374 			buf.append("stub[decline close request]");
375 			return buf;
376 		}
377 		public Object invoke(Invocation invocation) throws Throwable {
378 			CloseChannelRequest callback = (CloseChannelRequest) invocation.parameterValues.get(index);
379 			callback.reject();
380 			return null;
381 		}
382 	}
383 	
384 }