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 package org.apache.http.nio.client.methods;
28
29 import java.io.IOException;
30 import java.nio.ByteBuffer;
31 import java.nio.CharBuffer;
32 import java.util.concurrent.ExecutionException;
33 import java.util.concurrent.Future;
34 import java.util.concurrent.atomic.AtomicLong;
35
36 import org.apache.http.Consts;
37 import org.apache.http.localserver.HttpAsyncTestBase;
38 import org.apache.http.HttpException;
39 import org.apache.http.HttpHost;
40 import org.apache.http.HttpResponse;
41 import org.apache.http.entity.ContentType;
42 import org.apache.http.localserver.EchoHandler;
43 import org.apache.http.localserver.RandomHandler;
44 import org.apache.http.nio.IOControl;
45 import org.apache.http.nio.protocol.BasicAsyncRequestHandler;
46 import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
47 import org.apache.http.protocol.HttpContext;
48 import org.junit.Assert;
49 import org.junit.Before;
50 import org.junit.Test;
51 import org.mockito.Matchers;
52 import org.mockito.Mockito;
53
54 public class TestAsyncConsumers extends HttpAsyncTestBase {
55
56 @Before @Override
57 public void setUp() throws Exception {
58 super.setUp();
59 this.serverBootstrap.registerHandler("/echo/*", new BasicAsyncRequestHandler(new EchoHandler()));
60 this.serverBootstrap.registerHandler("/random/*", new BasicAsyncRequestHandler(new RandomHandler()));
61 }
62
63 static class ByteCountingConsumer extends AsyncByteConsumer<Long> {
64
65 public ByteCountingConsumer() {
66 super();
67 }
68
69 public ByteCountingConsumer(final int bufSize) {
70 super(bufSize);
71 }
72
73 private final AtomicLong count = new AtomicLong(0);
74
75 @Override
76 protected void onResponseReceived(final HttpResponse response) {
77 }
78
79 @Override
80 protected void onByteReceived(final ByteBuffer buf, final IOControl ioControl) {
81 this.count.addAndGet(buf.remaining());
82 }
83
84 @Override
85 protected Long buildResult(final HttpContext context) throws Exception {
86 return count.get();
87 }
88
89 }
90
91 @Test
92 public void testByteConsumer() throws Exception {
93 final HttpHost target = start();
94 for (int i = 0; i < 5; i++) {
95 final HttpAsyncRequestProducer httpget = HttpAsyncMethods.createGet(target.toURI() + "/random/20480");
96 final AsyncByteConsumer<Long> consumer = new ByteCountingConsumer();
97 final Future<Long> future = this.httpclient.execute(httpget, consumer, null);
98 final Long count = future.get();
99 Assert.assertEquals(20480, count.longValue());
100 }
101 }
102
103 @Test
104 public void testByteConsumerSmallBufffer() throws Exception {
105 final HttpHost target = start();
106 for (int i = 0; i < 5; i++) {
107 final HttpAsyncRequestProducer httpget = HttpAsyncMethods.createGet(target.toURI() + "/random/20480");
108 final AsyncByteConsumer<Long> consumer = new ByteCountingConsumer(512);
109 final Future<Long> future = this.httpclient.execute(httpget, consumer, null);
110 final Long count = future.get();
111 Assert.assertEquals(20480, count.longValue());
112 }
113 }
114
115 static class BufferingCharConsumer extends AsyncCharConsumer<String> {
116
117 public BufferingCharConsumer() {
118 super();
119 }
120
121 public BufferingCharConsumer(final int bufSize) {
122 super(bufSize);
123 }
124
125 private final StringBuilder sb = new StringBuilder();
126
127 @Override
128 public void onResponseReceived(final HttpResponse response) {
129 }
130
131 @Override
132 protected void onCharReceived(final CharBuffer buf, final IOControl ioControl) throws IOException {
133 while (buf.hasRemaining()) {
134 this.sb.append(buf.get());
135 }
136 }
137
138 @Override
139 protected void releaseResources() {
140 super.releaseResources();
141 this.sb.setLength(0);
142 }
143
144 @Override
145 protected String buildResult(final HttpContext context) throws Exception {
146 return this.sb.toString();
147 }
148
149 }
150
151 @Test
152 public void testCharConsumer() throws Exception {
153 final HttpHost target = start();
154 final StringBuilder sb = new StringBuilder();
155 for (int i= 0; i < 25; i++) {
156 sb.append("blah blah blah blah\r\n");
157 sb.append("yada yada yada yada\r\n");
158 }
159 final String s = sb.toString();
160
161 for (int i = 0; i < 5; i++) {
162 final HttpAsyncRequestProducer httppost = HttpAsyncMethods.createPost(
163 target.toURI() + "/echo/stuff", s,
164 ContentType.create("text/plain", Consts.ASCII));
165 final AsyncCharConsumer<String> consumer = new BufferingCharConsumer();
166 final Future<String> future = this.httpclient.execute(httppost, consumer, null);
167 final String result = future.get();
168 Assert.assertEquals(s, result);
169 }
170 }
171
172 @Test
173 public void testCharConsumerSmallBufffer() throws Exception {
174 final HttpHost target = start();
175 final StringBuilder sb = new StringBuilder();
176 for (int i= 0; i < 25; i++) {
177 sb.append("blah blah blah blah\r\n");
178 sb.append("yada yada yada yada\r\n");
179 }
180 final String s = sb.toString();
181
182 for (int i = 0; i < 5; i++) {
183 final HttpAsyncRequestProducer httppost = HttpAsyncMethods.createPost(
184 target.toURI() + "/echo/stuff", s,
185 ContentType.create("text/plain", Consts.ASCII));
186 final AsyncCharConsumer<String> consumer = new BufferingCharConsumer(512);
187 final Future<String> future = this.httpclient.execute(httppost, consumer, null);
188 final String result = future.get();
189 Assert.assertEquals(s, result);
190 }
191 }
192
193 @Test
194 public void testResourceReleaseOnSuccess() throws Exception {
195 final HttpHost target = start();
196 final StringBuilder sb = new StringBuilder();
197 for (int i= 0; i < 25; i++) {
198 sb.append("blah blah blah blah\r\n");
199 sb.append("yada yada yada yada\r\n");
200 }
201 final String s = sb.toString();
202
203 final HttpAsyncRequestProducer httppost = HttpAsyncMethods.createPost(
204 target.toURI() + "/echo/stuff", s,
205 ContentType.create("text/plain", Consts.ASCII));
206 final BufferingCharConsumer consumer = Mockito.spy(new BufferingCharConsumer());
207 final Future<String> future = this.httpclient.execute(httppost, consumer, null);
208 final String result = future.get();
209 Assert.assertEquals(s, result);
210 Mockito.verify(consumer).buildResult(Matchers.any(HttpContext.class));
211 Mockito.verify(consumer).releaseResources();
212 }
213
214 @Test
215 public void testResourceReleaseOnException() throws Exception {
216 final HttpHost target = start();
217 final HttpAsyncRequestProducer httppost = HttpAsyncMethods.createPost(
218 target.toURI() + "/echo/stuff", "stuff",
219 ContentType.create("text/plain", Consts.ASCII));
220 final AsyncCharConsumer<String> consumer = Mockito.spy(new BufferingCharConsumer());
221 Mockito.doThrow(new IOException("Kaboom")).when(consumer).onCharReceived(
222 Matchers.any(CharBuffer.class), Matchers.any(IOControl.class));
223
224 final Future<String> future = this.httpclient.execute(httppost, consumer, null);
225 try {
226 future.get();
227 Assert.fail("ExecutionException expected");
228 } catch (final ExecutionException ex) {
229 final Throwable t = ex.getCause();
230 Assert.assertNotNull(t);
231 Assert.assertTrue(t instanceof IOException);
232 Assert.assertEquals("Kaboom", t.getMessage());
233 }
234 Mockito.verify(consumer).releaseResources();
235 }
236
237 @Test
238 public void testResourceReleaseOnBuildFailure() throws Exception {
239 final HttpHost target = start();
240 final HttpAsyncRequestProducer httppost = HttpAsyncMethods.createPost(
241 target.toURI() + "/echo/stuff", "stuff",
242 ContentType.create("text/plain", Consts.ASCII));
243 final BufferingCharConsumer consumer = Mockito.spy(new BufferingCharConsumer());
244 Mockito.doThrow(new HttpException("Kaboom")).when(consumer).buildResult(Matchers.any(HttpContext.class));
245
246 final Future<String> future = this.httpclient.execute(httppost, consumer, null);
247 try {
248 future.get();
249 Assert.fail("ExecutionException expected");
250 } catch (final ExecutionException ex) {
251 final Throwable t = ex.getCause();
252 Assert.assertNotNull(t);
253 Assert.assertTrue(t instanceof HttpException);
254 Assert.assertEquals("Kaboom", t.getMessage());
255 }
256 Mockito.verify(consumer).releaseResources();
257 }
258
259 }