001: /*
002: * Licensed to the Apache Software Foundation (ASF) under one
003: * or more contributor license agreements. See the NOTICE file
004: * distributed with this work for additional information
005: * regarding copyright ownership. The ASF licenses this file
006: * to you under the Apache License, Version 2.0 (the
007: * "License"); you may not use this file except in compliance
008: * with the License. You may obtain a copy of the License at
009: *
010: * http://www.apache.org/licenses/LICENSE-2.0
011: *
012: * Unless required by applicable law or agreed to in writing,
013: * software distributed under the License is distributed on an
014: * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015: * KIND, either express or implied. See the License for the
016: * specific language governing permissions and limitations
017: * under the License.
018: */
019: package org.apache.axis2.transport.nhttp;
020:
021: import java.io.IOException;
022: import java.nio.ByteBuffer;
023: import java.nio.channels.Channels;
024: import java.nio.channels.ReadableByteChannel;
025: import java.nio.channels.WritableByteChannel;
026:
027: import org.apache.axiom.soap.SOAP11Constants;
028: import org.apache.axiom.soap.SOAP12Constants;
029: import org.apache.axiom.soap.SOAPFactory;
030: import org.apache.axis2.addressing.AddressingConstants;
031: import org.apache.axis2.context.ConfigurationContext;
032: import org.apache.axis2.context.MessageContext;
033: import org.apache.axis2.description.WSDL2Constants;
034: import org.apache.axis2.engine.MessageReceiver;
035: import org.apache.axis2.transport.nhttp.util.PipeImpl;
036: import org.apache.axis2.transport.nhttp.util.WorkerPool;
037: import org.apache.axis2.transport.nhttp.util.WorkerPoolFactory;
038: import org.apache.axis2.wsdl.WSDLConstants;
039: import org.apache.commons.logging.Log;
040: import org.apache.commons.logging.LogFactory;
041: import org.apache.http.ConnectionReuseStrategy;
042: import org.apache.http.Header;
043: import org.apache.http.HttpConnection;
044: import org.apache.http.HttpException;
045: import org.apache.http.HttpRequest;
046: import org.apache.http.HttpResponse;
047: import org.apache.http.HttpStatus;
048: import org.apache.http.HttpVersion;
049: import org.apache.http.entity.BasicHttpEntity;
050: import org.apache.http.impl.DefaultConnectionReuseStrategy;
051: import org.apache.http.nio.ContentDecoder;
052: import org.apache.http.nio.ContentEncoder;
053: import org.apache.http.nio.NHttpClientConnection;
054: import org.apache.http.nio.NHttpClientHandler;
055: import org.apache.http.params.HttpParams;
056: import org.apache.http.params.HttpParamsLinker;
057: import org.apache.http.protocol.BasicHttpProcessor;
058: import org.apache.http.protocol.HttpContext;
059: import org.apache.http.protocol.HttpExecutionContext;
060: import org.apache.http.protocol.HttpProcessor;
061: import org.apache.http.protocol.RequestConnControl;
062: import org.apache.http.protocol.RequestContent;
063: import org.apache.http.protocol.RequestExpectContinue;
064: import org.apache.http.protocol.RequestTargetHost;
065: import org.apache.http.protocol.RequestUserAgent;
066:
067: /**
068: * The client connection handler. An instance of this class is used by each IOReactor, to
069: * process every connection. Hence this class should not store any data related to a single
070: * connection - as this is being shared.
071: */
072: public class ClientHandler implements NHttpClientHandler {
073:
074: private static final Log log = LogFactory
075: .getLog(ClientHandler.class);
076:
077: /** the HTTP protocol parameters to adhere to for outgoing messages */
078: private final HttpParams params;
079: /** the HttpProcessor for response messages received */
080: private final HttpProcessor httpProcessor;
081: /** the connection re-use strategy */
082: private final ConnectionReuseStrategy connStrategy;
083:
084: /** the Axis2 configuration context */
085: ConfigurationContext cfgCtx = null;
086: /** the nhttp configuration */
087: private NHttpConfiguration cfg = null;
088:
089: private WorkerPool workerPool = null;
090:
091: private static final String REQUEST_BUFFER = "request-buffer";
092: private static final String RESPONSE_BUFFER = "response-buffer";
093: private static final String OUTGOING_MESSAGE_CONTEXT = "axis2_message_context";
094: private static final String REQUEST_SOURCE_CHANNEL = "request-source-channel";
095: private static final String RESPONSE_SINK_CHANNEL = "request-sink-channel";
096:
097: private static final String CONTENT_TYPE = "Content-Type";
098:
099: /**
100: * Create an instance of this client connection handler using the Axis2 configuration
101: * context and Http protocol parameters given
102: * @param cfgCtx the Axis2 configuration context
103: * @param params the Http protocol parameters to adhere to
104: */
105: public ClientHandler(final ConfigurationContext cfgCtx,
106: final HttpParams params) {
107: super ();
108: this .cfgCtx = cfgCtx;
109: this .params = params;
110: this .httpProcessor = getHttpProcessor();
111: this .connStrategy = new DefaultConnectionReuseStrategy();
112:
113: this .cfg = NHttpConfiguration.getInstance();
114: workerPool = WorkerPoolFactory.getWorkerPool(cfg
115: .getClientCoreThreads(), cfg.getClientMaxThreads(), cfg
116: .getClientKeepalive(), cfg.getClientQueueLen(),
117: "Client Worker thread group", "HttpClientWorker");
118: }
119:
120: public void requestReady(final NHttpClientConnection conn) {
121: // The connection is ready for submission of a new request
122: }
123:
124: /**
125: * Submit a new request over an already established connection, which has been
126: * 'kept alive'
127: * @param conn the connection to use to send the request, which has been kept open
128: * @param axis2Req the new request
129: */
130: public void submitRequest(final NHttpClientConnection conn,
131: Axis2HttpRequest axis2Req) {
132:
133: try {
134: HttpContext context = conn.getContext();
135:
136: context.setAttribute(HttpExecutionContext.HTTP_CONNECTION,
137: conn);
138: context.setAttribute(HttpExecutionContext.HTTP_TARGET_HOST,
139: axis2Req.getHttpHost());
140:
141: context.setAttribute(OUTGOING_MESSAGE_CONTEXT, axis2Req
142: .getMsgContext());
143: context.setAttribute(REQUEST_SOURCE_CHANNEL, axis2Req
144: .getSourceChannel());
145:
146: HttpRequest request = axis2Req.getRequest();
147: HttpParamsLinker.link(request, this .params);
148: this .httpProcessor.process(request, context);
149:
150: conn.submitRequest(request);
151: context.setAttribute(HttpExecutionContext.HTTP_REQUEST,
152: request);
153:
154: } catch (IOException e) {
155: handleException("I/O Error : " + e.getMessage(), e, conn);
156: } catch (HttpException e) {
157: handleException("Unexpected HTTP protocol error: "
158: + e.getMessage(), e, conn);
159: }
160: }
161:
162: /**
163: * Invoked when the destination is connected
164: * @param conn the connection being processed
165: * @param attachment the attachment set previously
166: */
167: public void connected(final NHttpClientConnection conn,
168: final Object attachment) {
169: try {
170: HttpContext context = conn.getContext();
171: Axis2HttpRequest axis2Req = (Axis2HttpRequest) attachment;
172:
173: context.setAttribute(HttpExecutionContext.HTTP_CONNECTION,
174: conn);
175: context.setAttribute(HttpExecutionContext.HTTP_TARGET_HOST,
176: axis2Req.getHttpHost());
177:
178: // allocate temporary buffers to process this request
179: context.setAttribute(REQUEST_BUFFER, ByteBuffer
180: .allocate(cfg.getBufferZise()));
181: context.setAttribute(RESPONSE_BUFFER, ByteBuffer
182: .allocate(cfg.getBufferZise()));
183:
184: context.setAttribute(OUTGOING_MESSAGE_CONTEXT, axis2Req
185: .getMsgContext());
186: context.setAttribute(REQUEST_SOURCE_CHANNEL, axis2Req
187: .getSourceChannel());
188:
189: HttpRequest request = axis2Req.getRequest();
190: HttpParamsLinker.link(request, this .params);
191: this .httpProcessor.process(request, context);
192:
193: conn.submitRequest(request);
194: context.setAttribute(HttpExecutionContext.HTTP_REQUEST,
195: request);
196:
197: } catch (IOException e) {
198: handleException("I/O Error : " + e.getMessage(), e, conn);
199: } catch (HttpException e) {
200: handleException("Unexpected HTTP protocol error: "
201: + e.getMessage(), e, conn);
202: }
203: }
204:
205: public void closed(final NHttpClientConnection conn) {
206: log.trace("Connection closed");
207: }
208:
209: /**
210: * Handle connection timeouts by shutting down the connections
211: * @param conn the connection being processed
212: */
213: public void timeout(final NHttpClientConnection conn) {
214: log.debug("Connection Timeout");
215: shutdownConnection(conn);
216: }
217:
218: /**
219: * Handle Http protocol violations encountered while reading from underlying channels
220: * @param conn the connection being processed
221: * @param e the exception encountered
222: */
223: public void exception(final NHttpClientConnection conn,
224: final HttpException e) {
225: log.error("HTTP protocol violation : " + e.getMessage());
226: shutdownConnection(conn);
227: }
228:
229: /**
230: * Handle IO errors while reading or writing to underlying channels
231: * @param conn the connection being processed
232: * @param e the exception encountered
233: */
234: public void exception(final NHttpClientConnection conn,
235: final IOException e) {
236: log.error("I/O error : " + e.getMessage(), e);
237: shutdownConnection(conn);
238: }
239:
240: /**
241: * Process ready input (i.e. response from remote server)
242: * @param conn connection being processed
243: * @param decoder the content decoder in use
244: */
245: public void inputReady(final NHttpClientConnection conn,
246: final ContentDecoder decoder) {
247: HttpContext context = conn.getContext();
248: HttpResponse response = conn.getHttpResponse();
249: WritableByteChannel sink = (WritableByteChannel) context
250: .getAttribute(RESPONSE_SINK_CHANNEL);
251: ByteBuffer inbuf = (ByteBuffer) context
252: .getAttribute(REQUEST_BUFFER);
253:
254: try {
255: while (decoder.read(inbuf) > 0) {
256: inbuf.flip();
257: sink.write(inbuf);
258: inbuf.compact();
259: }
260:
261: if (decoder.isCompleted()) {
262: if (sink != null)
263: sink.close();
264: if (!connStrategy.keepAlive(response, context)) {
265: conn.close();
266: } else {
267: ConnectionPool.release(conn);
268: }
269: }
270:
271: } catch (IOException e) {
272: handleException("I/O Error : " + e.getMessage(), e, conn);
273: }
274: }
275:
276: /**
277: * Process ready output (i.e. write request to remote server)
278: * @param conn the connection being processed
279: * @param encoder the encoder in use
280: */
281: public void outputReady(final NHttpClientConnection conn,
282: final ContentEncoder encoder) {
283: HttpContext context = conn.getContext();
284:
285: ReadableByteChannel source = (ReadableByteChannel) context
286: .getAttribute(REQUEST_SOURCE_CHANNEL);
287: ByteBuffer outbuf = (ByteBuffer) context
288: .getAttribute(RESPONSE_BUFFER);
289:
290: try {
291: int bytesRead = source.read(outbuf);
292: if (bytesRead == -1) {
293: encoder.complete();
294: } else {
295: outbuf.flip();
296: encoder.write(outbuf);
297: outbuf.compact();
298: }
299:
300: if (encoder.isCompleted()) {
301: source.close();
302: }
303:
304: } catch (IOException e) {
305: handleException("I/O Error : " + e.getMessage(), e, conn);
306: }
307: }
308:
309: /**
310: * Process a response received for the request sent out
311: * @param conn the connection being processed
312: */
313: public void responseReceived(final NHttpClientConnection conn) {
314: HttpContext context = conn.getContext();
315: HttpResponse response = conn.getHttpResponse();
316:
317: switch (response.getStatusLine().getStatusCode()) {
318: case HttpStatus.SC_ACCEPTED: {
319: log.debug("Received a 202 Accepted response");
320:
321: // create a dummy message with an empty SOAP envelope and a property
322: // NhttpConstants.SC_ACCEPTED set to Boolean.TRUE to indicate this is a
323: // placeholder message for the transport to send a HTTP 202 to the
324: // client. Should / would be ignored by any transport other than
325: // nhttp. For example, JMS would not send a reply message for one-way
326: // operations.
327: MessageContext outMsgCtx = (MessageContext) context
328: .getAttribute(OUTGOING_MESSAGE_CONTEXT);
329: MessageReceiver mr = outMsgCtx.getAxisOperation()
330: .getMessageReceiver();
331:
332: try {
333: MessageContext responseMsgCtx = outMsgCtx
334: .getOperationContext().getMessageContext(
335: WSDL2Constants.MESSAGE_LABEL_IN);
336: if (responseMsgCtx == null) {
337: // to support Sandesha.. however, this means that we received a 202 accepted
338: // for an out-only , for which we do not need a dummy message anyway
339: return;
340: }
341: responseMsgCtx.setServerSide(true);
342: responseMsgCtx.setDoingREST(outMsgCtx.isDoingREST());
343: responseMsgCtx
344: .setProperty(
345: MessageContext.TRANSPORT_IN,
346: outMsgCtx
347: .getProperty(MessageContext.TRANSPORT_IN));
348: responseMsgCtx.setTransportIn(outMsgCtx
349: .getTransportIn());
350: responseMsgCtx.setTransportOut(outMsgCtx
351: .getTransportOut());
352:
353: responseMsgCtx.setAxisMessage(outMsgCtx
354: .getAxisOperation().getMessage(
355: WSDLConstants.MESSAGE_LABEL_IN_VALUE));
356: responseMsgCtx.setOperationContext(outMsgCtx
357: .getOperationContext());
358: responseMsgCtx.setConfigurationContext(outMsgCtx
359: .getConfigurationContext());
360: responseMsgCtx.setTo(null);
361:
362: responseMsgCtx.setEnvelope(((SOAPFactory) outMsgCtx
363: .getEnvelope().getOMFactory())
364: .getDefaultEnvelope());
365: responseMsgCtx
366: .setProperty(
367: AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES,
368: Boolean.TRUE);
369: responseMsgCtx.setProperty(NhttpConstants.SC_ACCEPTED,
370: Boolean.TRUE);
371: mr.receive(responseMsgCtx);
372:
373: } catch (org.apache.axis2.AxisFault af) {
374: log
375: .error(
376: "Unable to report back 202 Accepted state to the message receiver",
377: af);
378: }
379:
380: return;
381: }
382: case HttpStatus.SC_INTERNAL_SERVER_ERROR: {
383: Header contentType = response.getFirstHeader(CONTENT_TYPE);
384: if (contentType != null
385: && (contentType.getValue().indexOf(
386: SOAP11Constants.SOAP_11_CONTENT_TYPE) >= 0)
387: || contentType.getValue().indexOf(
388: SOAP12Constants.SOAP_12_CONTENT_TYPE) >= 0) {
389: log
390: .debug("Received an internal server error with a SOAP payload");
391: processResponse(conn, context, response);
392: return;
393: }
394: log.error("Received an internal server error : "
395: + response.getStatusLine().getReasonPhrase());
396: return;
397: }
398: case HttpStatus.SC_OK: {
399: processResponse(conn, context, response);
400: }
401: }
402: }
403:
404: /**
405: * Perform processing of the received response though Axis2
406: * @param conn
407: * @param context
408: * @param response
409: */
410: private void processResponse(final NHttpClientConnection conn,
411: HttpContext context, HttpResponse response) {
412:
413: try {
414: PipeImpl responsePipe = new PipeImpl();
415: context.setAttribute(RESPONSE_SINK_CHANNEL, responsePipe
416: .sink());
417:
418: BasicHttpEntity entity = new BasicHttpEntity();
419: if (response.getStatusLine().getHttpVersion()
420: .greaterEquals(HttpVersion.HTTP_1_1)) {
421: entity.setChunked(true);
422: }
423: response.setEntity(entity);
424: context.setAttribute(HttpContext.HTTP_RESPONSE, response);
425:
426: workerPool.execute(new ClientWorker(cfgCtx, Channels
427: .newInputStream(responsePipe.source()), response,
428: (MessageContext) context
429: .getAttribute(OUTGOING_MESSAGE_CONTEXT)));
430:
431: } catch (IOException e) {
432: handleException("I/O Error : " + e.getMessage(), e, conn);
433: }
434: }
435:
436: // ----------- utility methods -----------
437:
438: private void handleException(String msg, Exception e,
439: NHttpClientConnection conn) {
440: log.error(msg, e);
441: if (conn != null) {
442: shutdownConnection(conn);
443: }
444: }
445:
446: /**
447: * Shutdown the connection ignoring any IO errors during the process
448: * @param conn the connection to be shutdown
449: */
450: private void shutdownConnection(final HttpConnection conn) {
451: try {
452: conn.shutdown();
453: } catch (IOException ignore) {
454: }
455: }
456:
457: /**
458: * Return the HttpProcessor for requests
459: * @return the HttpProcessor that processes requests
460: */
461: private HttpProcessor getHttpProcessor() {
462: BasicHttpProcessor httpProcessor = new BasicHttpProcessor();
463: httpProcessor.addInterceptor(new RequestContent());
464: httpProcessor.addInterceptor(new RequestTargetHost());
465: httpProcessor.addInterceptor(new RequestConnControl());
466: httpProcessor.addInterceptor(new RequestUserAgent());
467: httpProcessor.addInterceptor(new RequestExpectContinue());
468: return httpProcessor;
469: }
470:
471: }
|