You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by hi...@apache.org on 2013/07/12 00:18:48 UTC
svn commit: r1502386 [6/7] - in /synapse/trunk/java:
modules/core/src/main/java/org/apache/synapse/
modules/core/src/main/java/org/apache/synapse/config/xml/
modules/core/src/main/java/org/apache/synapse/config/xml/endpoints/
modules/core/src/main/java...
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/config/TargetConfiguration.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/config/TargetConfiguration.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/config/TargetConfiguration.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/config/TargetConfiguration.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.config;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.description.ParameterInclude;
+import org.apache.axis2.transport.base.threads.WorkerPool;
+import org.apache.http.HttpRequestInterceptor;
+import org.apache.http.params.HttpParams;
+import org.apache.http.protocol.*;
+import org.apache.synapse.transport.passthru.connections.TargetConnections;
+
+/**
+ * This class stores configuration specific to HTTP Connectors (Senders)
+ */
+public class TargetConfiguration extends BaseConfiguration {
+
+ private HttpProcessor httpProcessor = null;
+
+ private int maxConnections = Integer.MAX_VALUE;
+
+ /** Weather User-Agent header coming from client should be preserved */
+ private boolean preserveUserAgentHeader = false;
+ /** Weather Server header coming from server should be preserved */
+ private boolean preserveServerHeader = true;
+
+ private TargetConnections connections = null;
+
+ public TargetConfiguration(ConfigurationContext configurationContext,
+ ParameterInclude parameters,
+ WorkerPool pool) {
+ super(configurationContext, parameters, pool);
+
+ httpProcessor = new ImmutableHttpProcessor(
+ new HttpRequestInterceptor[] {
+ new RequestContent(),
+ new RequestTargetHost(),
+ new RequestConnControl(),
+ new RequestUserAgent(),
+ new RequestExpectContinue()
+ });
+ }
+
+ public void build() throws AxisFault {
+ super.build();
+
+ maxConnections = conf.getIntProperty(PassThroughConfigPNames.MAX_CONNECTION_PER_HOST_PORT,
+ Integer.MAX_VALUE);
+ preserveUserAgentHeader = conf.isPreserveUserAgentHeader();
+ preserveServerHeader = conf.isPreserveServerHeader();
+ }
+
+ public HttpParams getHttpParameters() {
+ return httpParameters;
+ }
+
+ public HttpProcessor getHttpProcessor() {
+ return httpProcessor;
+ }
+
+ public int getMaxConnections() {
+ return maxConnections;
+ }
+
+ public boolean isPreserveUserAgentHeader() {
+ return preserveUserAgentHeader;
+ }
+
+ public boolean isPreserveServerHeader() {
+ return preserveServerHeader;
+ }
+
+ public TargetConnections getConnections() {
+ return connections;
+ }
+
+ public void setConnections(TargetConnections connections) {
+ this.connections = connections;
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/HostConnections.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/HostConnections.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/HostConnections.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/HostConnections.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,160 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.connections;
+
+import org.apache.http.nio.NHttpClientConnection;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.protocol.ExecutionContext;
+import org.apache.http.protocol.HttpContext;
+
+import java.util.List;
+import java.util.ArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * This stores connections for a particular host + port.
+ */
+public class HostConnections {
+ private static final Log log = LogFactory.getLog(HostConnections.class);
+ // host
+ private String host;
+ // port
+ private int port;
+ // maximum number of connections allowed for this host + port
+ private int maxSize;
+ // number of awaiting connections
+ private int pendingConnections;
+ // list of free connections available
+ private List<NHttpClientConnection> freeConnections = new ArrayList<NHttpClientConnection>();
+ // list of connections in use
+ private List<NHttpClientConnection> busyConnections = new ArrayList<NHttpClientConnection>();
+
+ private Lock lock = new ReentrantLock();
+
+ public HostConnections(String host, int port, int maxSize) {
+ if (log.isDebugEnabled()) {
+ log.debug("Creating new connection pool to the host: " + host + ", port: " + port);
+ }
+ this.host = host;
+ this.port = port;
+ this.maxSize = maxSize;
+ }
+
+ /**
+ * Get a connection for the host:port
+ *
+ * @return a connection
+ */
+ public NHttpClientConnection getConnection() {
+ lock.lock();
+ try {
+ if (freeConnections.size() > 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("Returning an existing free connection to " + host + ":" + port);
+ }
+ NHttpClientConnection conn = freeConnections.get(0);
+ freeConnections.remove(conn);
+ busyConnections.add(conn);
+ return conn;
+ }
+ } finally {
+ lock.unlock();
+ }
+ return null;
+ }
+
+ public void release(NHttpClientConnection conn) {
+ conn.getMetrics().reset();
+ HttpContext ctx = conn.getContext();
+ ctx.removeAttribute(ExecutionContext.HTTP_REQUEST);
+ ctx.removeAttribute(ExecutionContext.HTTP_RESPONSE);
+
+ lock.lock();
+ try {
+ if (busyConnections.remove(conn)) {
+ freeConnections.add(conn);
+ } else {
+ log.error("Attempted to releaseConnection connection not in the busy list");
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void forget(NHttpClientConnection conn) {
+ lock.lock();
+ try {
+ if (!freeConnections.remove(conn)) {
+ busyConnections.remove(conn);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void addConnection(NHttpClientConnection conn) {
+ if (log.isDebugEnabled()) {
+ log.debug("New connection to " + host + ":" + port + " is added to the free list");
+ }
+ lock.lock();
+ try {
+ freeConnections.add(conn);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Indicates that a connection has been successfully established with a remote server
+ * as notified by the session request call back.
+ */
+ public synchronized void pendingConnectionSucceeded() {
+ lock.lock();
+ try {
+ pendingConnections--;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Keep track of the number of times connections to this host:port has failed
+ * consecutively
+ */
+ public void pendingConnectionFailed() {
+ lock.lock();
+ try {
+ pendingConnections--;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public boolean canHaveMoreConnections() {
+ return busyConnections.size() + pendingConnections < maxSize;
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/SourceConnections.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/SourceConnections.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/SourceConnections.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/SourceConnections.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,175 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.synapse.transport.passthru.connections;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.nio.NHttpServerConnection;
+import org.apache.synapse.transport.passthru.SourceContext;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Keeps track of the connections coming in to the transport.
+ */
+public class SourceConnections {
+ private static Log log = LogFactory.getLog(SourceConnections.class);
+
+ /** The pool of connections in use */
+ private volatile List<NHttpServerConnection> busyConnections = new ArrayList<NHttpServerConnection>();
+
+ /** The pool of connections that are not being used */
+ private volatile List<NHttpServerConnection> freeConnections = new ArrayList<NHttpServerConnection>();
+
+ /** Lock for synchronizing the access to the pools */
+ private final Lock lock = new ReentrantLock();
+
+ /** Max free connections allowed in a given instance of time */
+ private int maxFreeConnections;
+
+ /** Max busy connections allowed in a given instance of time */
+ private int maxBusyConnections;
+
+ /**
+ * Add a connection to the pool.
+ *
+ * @param conn connection to be added
+ */
+ public void addConnection(NHttpServerConnection conn) {
+ lock.lock();
+ try {
+ freeConnections.add(conn);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * This method should be called when ever a connection being used for processing
+ * a request-response.
+ *
+ * @param conn the connection to be used
+ */
+ public void useConnection(NHttpServerConnection conn) {
+ lock.lock();
+ try {
+ boolean freeConnection = freeConnections.remove(conn);
+ if (freeConnection) {
+ busyConnections.add(conn);
+ } else {
+ if (busyConnections.contains(conn)) {
+ throw new IllegalStateException("The connection is busy. " +
+ "Cannot use it for new request");
+ } else {
+ throw new IllegalStateException("Trying to use a connection " +
+ "which is not in free connections " + conn);
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * This method should be called after a connection is being used for a request-response.
+ *
+ * @param conn the connection being used
+ */
+ public void releaseConnection(NHttpServerConnection conn) {
+ lock.lock();
+ try {
+ SourceContext.get(conn).reset();
+
+ if (busyConnections.remove(conn)) {
+ freeConnections.add(conn);
+ } else {
+ throw new IllegalStateException("Trying to finish using a connection " +
+ "which is not in busy connections " + conn);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Shutdown a connection
+ *
+ * @param conn the connection that needs to be shut down
+ */
+ public void shutDownConnection(NHttpServerConnection conn) {
+ if (log.isDebugEnabled()) {
+ log.debug("Shutting down connection forcefully " + conn);
+ }
+ lock.lock();
+ try {
+ SourceContext.get(conn).reset();
+
+ if (!busyConnections.remove(conn)) {
+ freeConnections.remove(conn);
+ }
+
+ try {
+ conn.shutdown();
+ } catch (IOException ignored) {
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+
+ /**
+ * Close a connection gracefully.
+ *
+ * @param conn the connection that needs to be closed.
+ */
+ public void closeConnection(NHttpServerConnection conn) {
+ if (log.isDebugEnabled()) {
+ log.debug("Shutting down connection forcefully " + conn);
+ }
+ lock.lock();
+ try {
+ SourceContext.get(conn).reset();
+
+ if (!busyConnections.remove(conn)) {
+ freeConnections.remove(conn);
+ }
+
+ try {
+ conn.close();
+ } catch (IOException ignored) {
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void destroy() {
+ for (NHttpServerConnection conn : freeConnections) {
+ shutDownConnection(conn);
+ }
+
+ // for all the busy connections we have to notify that their cannot
+ // be anymore requests over them
+ for (NHttpServerConnection conn : busyConnections) {
+ SourceContext.get(conn).setShutDown(true);
+ }
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/TargetConnections.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/TargetConnections.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/TargetConnections.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/connections/TargetConnections.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,176 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.connections;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.nio.NHttpClientConnection;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.synapse.transport.passthru.ConnectCallback;
+import org.apache.synapse.transport.passthru.PassThroughConstants;
+import org.apache.synapse.transport.passthru.TargetContext;
+import org.apache.synapse.transport.passthru.config.TargetConfiguration;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.net.InetSocketAddress;
+
+/**
+ * Manages the connection from transport to the back end servers. It keeps track of the
+ * connections for host:port pair.
+ */
+public class TargetConnections {
+ private static final Log log = LogFactory.getLog(TargetConnections.class);
+
+ /** map to hold the ConnectionPools. The key is host:port */
+ private final Map<String, HostConnections> poolMap =
+ new ConcurrentHashMap<String, HostConnections>();
+
+ /** max connections per host:port pair. At the moment all the host:ports can
+ * have the same max */
+ private int maxConnections;
+
+ /** io-reactor to use for creating connections */
+ private ConnectingIOReactor ioReactor;
+
+ /** callback invoked when a connection is made */
+ private ConnectCallback callback = null;
+
+ /**
+ * Create a TargetConnections with the given IO-Reactor
+ *
+ * @param ioReactor the IO-Reactor
+ * @param targetConfiguration the configuration of the sender
+ * @param callback the callback
+ */
+ public TargetConnections(ConnectingIOReactor ioReactor,
+ TargetConfiguration targetConfiguration,
+ ConnectCallback callback) {
+
+ this.maxConnections = targetConfiguration.getMaxConnections();
+ this.ioReactor = ioReactor;
+ this.callback = callback;
+ }
+
+ /**
+ * Return a connection to the host:port pair. If a connection is not available
+ * return <code>null</code>. If the particular host:port allows to create more connections
+ * this method will try to connect asynchronously. If the connection is successful it will
+ * be notified in a separate thread.
+ *
+ * @param host host
+ * @param port port
+ * @return Either returns a connection if already available or returns null and notifies
+ * the delivery agent when the connection is available
+ */
+ public NHttpClientConnection getConnection(String host, int port) {
+ if (log.isDebugEnabled()) {
+ log.debug("Trying to get a connection to the " + host + ":" + port);
+ }
+
+ HostConnections pool = getConnectionPool(host, port);
+
+ // trying to get an existing connection
+ NHttpClientConnection connection = pool.getConnection();
+ if (connection == null) {
+ if (pool.canHaveMoreConnections()) {
+ ioReactor.connect(new InetSocketAddress(host, port), null, pool, callback);
+ } else {
+ log.warn("Connection pool reached maximum allowed connections for: "
+ + host + ":" + port + ". Target server may have become slow");
+ }
+ }
+
+ return connection;
+ }
+
+ /**
+ * This connection is no longer valid. So we need to shutdownConnection connection.
+ *
+ * @param conn connection to shutdownConnection
+ */
+ public void shutdownConnection(NHttpClientConnection conn) {
+ HostConnections pool = (HostConnections) conn.getContext().getAttribute(
+ PassThroughConstants.CONNECTION_POOL);
+
+ TargetContext.get(conn).reset();
+
+ if (pool != null) {
+ pool.forget(conn);
+ } else {
+ // we shouldn't get here
+ log.fatal("Connection without a pool. Something wrong. Need to fix.");
+ }
+
+ try {
+ conn.shutdown();
+ } catch (IOException ignored) {
+ }
+ }
+
+ /**
+ * Release an active connection to the pool
+ *
+ * @param conn connection to be released
+ */
+ public void releaseConnection(NHttpClientConnection conn) {
+ HostConnections pool = (HostConnections) conn.getContext().getAttribute(
+ PassThroughConstants.CONNECTION_POOL);
+
+ TargetContext.get(conn).reset();
+
+ if (pool != null) {
+ pool.release(conn);
+ } else {
+ // we shouldn't get here
+ log.fatal("Connection without a pool. Something wrong. Need to fix.");
+ }
+ }
+
+ /**
+ * This method is called when a new connection is made.
+ *
+ * @param conn connection to the target server
+ */
+ public void addConnection(NHttpClientConnection conn) {
+ HostConnections pool = (HostConnections) conn.getContext().getAttribute(
+ PassThroughConstants.CONNECTION_POOL);
+ if (pool != null) {
+ pool.addConnection(conn);
+ } else {
+ // we shouldn't get here
+ log.fatal("Connection without a pool. Something wrong. Need to fix.");
+ }
+ }
+
+ private HostConnections getConnectionPool(String host, int port) {
+ String key = host + ":" + port;
+ // see weather a pool already exists for this host:port
+ HostConnections pool = poolMap.get(key);
+ synchronized (poolMap) {
+ if (pool == null) {
+ pool = new HostConnections(host, port, maxConnections);
+ poolMap.put(key, pool);
+ }
+ }
+
+ return pool;
+ }
+
+
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsView.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsView.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsView.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsView.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,268 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import org.apache.axis2.AxisFault;
+
+import java.util.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * ConnectionsView MBean can be used to collect and monitor statistics on HTTP connections
+ * created by the NHTTP transport. Connection statistics can be divided into two categories,
+ * namely short term data and long term data. Short term data is related to the last 15
+ * minutes of execution and they are updated every minute. Long term data is related to
+ * the last 24 hours of execution and they get updated every 5 minutes. In addition to the
+ * connection statistics this MBean also provides information on the request and response
+ * sizes received over the HTTP connections. All messages are divided into six categories
+ * based on their sizes and the resulting counts are made available as a table.
+ */
+public class ConnectionsView implements ConnectionsViewMBean {
+
+ private static final String PASS_THROUGH_CONNECTIONS = "PassThroughConnections";
+
+ // Bucket definitions
+ private static final int LESS_THAN_1K = 0;
+ private static final int LESS_THAN_10K = 1;
+ private static final int LESS_THAN_100K = 2;
+ private static final int LESS_THAN_1M = 3;
+ private static final int LESS_THAN_10M = 4;
+ private static final int GREATER_THAN_10M = 5;
+
+ private static final int SHORT_DATA_COLLECTION_PERIOD = 60;
+ private static final int LONG_DATA_COLLECTION_PERIOD = 60 * 5;
+
+ private static final int SAMPLES_PER_HOUR = (60 * 60)/LONG_DATA_COLLECTION_PERIOD;
+
+ private Queue<Integer> shortTermDataQueue = new LinkedList<Integer>();
+ private Queue<Integer> longTermDataQueue = new LinkedList<Integer>();
+
+ private AtomicInteger activeConnections = new AtomicInteger(0);
+ private AtomicInteger shortTermOpenedConnections = new AtomicInteger(0);
+ private AtomicInteger longTermOpenedConnections = new AtomicInteger(0);
+
+ // The array length must be equal to the number of buckets
+ private AtomicInteger[] requestSizeCounters = new AtomicInteger[6];
+ private AtomicInteger[] responseSizeCounters = new AtomicInteger[6];
+
+ private Date resetTime = Calendar.getInstance().getTime();
+
+ private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
+ private String name;
+
+ public ConnectionsView(String name) throws AxisFault {
+ this.name = name;
+
+ initCounters(requestSizeCounters);
+ initCounters(responseSizeCounters);
+
+ Runnable task = new Runnable() {
+ public void run() {
+ // We only need historical data for the last 15 minutes
+ // Therefore no need to keep data older than that...
+ if (shortTermDataQueue.size() == 15) {
+ shortTermDataQueue.remove();
+ }
+ shortTermDataQueue.offer(shortTermOpenedConnections.getAndSet(0));
+ }
+ };
+ // Delay the timer by 1 minute to prevent the task from starting immediately
+ scheduler.scheduleAtFixedRate(task, SHORT_DATA_COLLECTION_PERIOD,
+ SHORT_DATA_COLLECTION_PERIOD, TimeUnit.SECONDS);
+
+ Runnable longTermCollector = new Runnable() {
+ public void run() {
+ // We only need historical data for the last 24 hours
+ // Therefore no need to keep data older than that...
+ if (longTermDataQueue.size() == 24 * SAMPLES_PER_HOUR) {
+ longTermDataQueue.remove();
+ }
+ longTermDataQueue.offer(longTermOpenedConnections.getAndSet(0));
+ }
+ };
+ scheduler.scheduleAtFixedRate(longTermCollector, LONG_DATA_COLLECTION_PERIOD,
+ LONG_DATA_COLLECTION_PERIOD, TimeUnit.SECONDS);
+
+ MBeanRegistrar.getInstance().registerMBean(this, PASS_THROUGH_CONNECTIONS, name);
+
+ }
+
+ public void destroy() {
+ MBeanRegistrar.getInstance().unRegisterMBean(PASS_THROUGH_CONNECTIONS, name);
+ scheduler.shutdownNow();
+ }
+
+ private void initCounters(AtomicInteger[] counters) {
+ for (int i = 0; i < counters.length; i++) {
+ if (counters[i] == null) {
+ counters[i] = new AtomicInteger(0);
+ } else {
+ counters[i].set(0);
+ }
+ }
+ }
+
+ protected void connected() {
+ activeConnections.incrementAndGet();
+ shortTermOpenedConnections.incrementAndGet();
+ longTermOpenedConnections.incrementAndGet();
+ }
+
+ protected void disconnected() {
+ activeConnections.decrementAndGet();
+ }
+
+ protected void notifyMessageSize(long size, boolean isRequest) {
+ // This logic gets executed for each and every transaction. For a typical
+ // mediation scenario this method will be called 4 times. Therefore I'm using
+ // arrays of integers to keep the overhead down to a minimum. Since the number
+ // of buckets is 6, this can be easily managed without using a slow data structure
+ // like a HashMap. This approach guarantees O(1) complexity.
+
+ AtomicInteger[] counters = isRequest ? requestSizeCounters : responseSizeCounters;
+
+ if (size < 1024) {
+ counters[LESS_THAN_1K].incrementAndGet();
+ } else if (size < 10240) {
+ counters[LESS_THAN_10K].incrementAndGet();
+ } else if (size < 102400) {
+ counters[LESS_THAN_100K].incrementAndGet();
+ } else if (size < 1048576) {
+ counters[LESS_THAN_1M].incrementAndGet();
+ } else if (size < 10485760) {
+ counters[LESS_THAN_10M].incrementAndGet();
+ } else {
+ counters[GREATER_THAN_10M].incrementAndGet();
+ }
+ }
+
+ public int getActiveConnections() {
+ return activeConnections.get();
+ }
+
+ public int getLastMinuteConnections() {
+ return getTotalConnections(1);
+ }
+
+ public int getLast5MinuteConnections() {
+ return getTotalConnections(5);
+ }
+
+ public int getLast15MinuteConnections() {
+ return getTotalConnections(15);
+ }
+
+ public int getLastHourConnections() {
+ return getTotalConnectionsByHour(1);
+ }
+
+ public int getLast8HourConnections() {
+ return getTotalConnectionsByHour(8);
+ }
+
+ public int getLast24HourConnections() {
+ return getTotalConnectionsByHour(24);
+ }
+
+ public Map getRequestSizesMap() {
+ return getCountersMap(requestSizeCounters);
+ }
+
+ public Map getResponseSizesMap() {
+ return getCountersMap(responseSizeCounters);
+ }
+
+ public Date getLastResetTime() {
+ return resetTime;
+ }
+
+ private Map<String,Integer> getCountersMap(AtomicInteger[] counters) {
+ // This ensures that keys are returned in the same order we insert them
+ // Provides better readability in the JMX consoles
+ Map<String,Integer> map = new LinkedHashMap<String,Integer>();
+ map.put("< 1 K", counters[LESS_THAN_1K].get());
+ map.put("< 10 K", counters[LESS_THAN_10K].get());
+ map.put("< 100 K", counters[LESS_THAN_100K].get());
+ map.put("< 1 M", counters[LESS_THAN_1M].get());
+ map.put("< 10 M", counters[LESS_THAN_10M].get());
+ map.put("> 10 M", counters[GREATER_THAN_10M].get());
+ return map;
+ }
+
+ public void reset() {
+ initCounters(requestSizeCounters);
+ initCounters(responseSizeCounters);
+ shortTermDataQueue.clear();
+ longTermDataQueue.clear();
+ resetTime = Calendar.getInstance().getTime();
+ }
+
+ /**
+ * Return the number of total connections opened during last 'n' munites
+ * of execution
+ *
+ * @param n Number of minutes in the execution history
+ * @return The number of connections opened
+ */
+ private int getTotalConnections(int n) {
+ int sum = 0;
+ Integer[] array = shortTermDataQueue.toArray(new Integer[shortTermDataQueue.size()]);
+
+ if (n > array.length) {
+ for (int i = 0; i < array.length; i++) {
+ sum += array[i];
+ }
+ } else {
+ for (int i = 0; i < n; i++) {
+ sum += array[array.length - 1 - i];
+ }
+ }
+ return sum;
+ }
+
+ /**
+ * Return the number of total connections opened during last 'n' hours
+ * of execution
+ *
+ * @param n Number of hours in the execution history
+ * @return The number of connections opened
+ */
+ private int getTotalConnectionsByHour(int n) {
+ int samples = n * SAMPLES_PER_HOUR;
+ int sum = 0;
+ Integer[] array = longTermDataQueue.toArray(new Integer[longTermDataQueue.size()]);
+
+ if (samples > array.length) {
+ for (int i = 0; i < array.length; i++) {
+ sum += array[i];
+ }
+ } else {
+ for (int i = 0; i < samples; i++) {
+ sum += array[array.length - 1 - i];
+ }
+ }
+ return sum;
+ }
+}
+
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsViewMBean.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsViewMBean.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsViewMBean.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/ConnectionsViewMBean.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,41 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import java.util.Date;
+import java.util.Map;
+
+public interface ConnectionsViewMBean {
+
+ public int getActiveConnections();
+ public int getLastMinuteConnections();
+ public int getLast5MinuteConnections();
+ public int getLast15MinuteConnections();
+ public int getLastHourConnections();
+ public int getLast8HourConnections();
+ public int getLast24HourConnections();
+ public Map getRequestSizesMap();
+ public Map getResponseSizesMap();
+ public Date getLastResetTime();
+
+ public void reset();
+
+}
+
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,260 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import org.apache.axis2.AxisFault;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * <p>LatencyView provides statistical information related to the latency (overhead) incurred by
+ * the Synapse NHTTP transport, when mediating messages back and forth. Statistics are available
+ * under two main categories, namely short term data and long term data. Short term data is
+ * statistical information related to the last 15 minutes of execution and these metrics are
+ * updated every 5 seconds. Long term data is related to the last 24 hours of execution and
+ * they are updated every 5 minutes. Two timer tasks and a single threaded scheduled executor
+ * is used to perform these periodic calculations.</p>
+ *
+ * <p>Latency calculation for a single invocation is carried out by taking timestamps on
+ * following events:</p>
+ *
+ * <ul>
+ * <li>t1 - Receiving a new request (ServerHandler#requestReceived)</li>
+ * <li>t2 - Obtaining a connection to forward the request (Clienthandler#processConnection)</li>
+ * <li>t3 - Reading the complete response from the backend server (ClientHandler#inputReady)</li>
+ * <li>t4 - Writing the complete response to the client (ServerHandler#outputReady)</li>
+ * <ul>
+ *
+ * <p>Having taken these timestamps, the latency for the invocation is calculated as follows:<br/>
+ * Latency = (t4 - t1) - (t3 - t2)
+ * </p>
+ *
+ */
+public class LatencyView implements LatencyViewMBean {
+
+ private static final String NHTTP_LATENCY_VIEW = "NhttpTransportLatency";
+
+ private static final int SMALL_DATA_COLLECTION_PERIOD = 5;
+ private static final int LARGE_DATA_COLLECTION_PERIOD = 5 * 60;
+ private static final int SAMPLES_PER_MINUTE = 60/ SMALL_DATA_COLLECTION_PERIOD;
+ private static final int SAMPLES_PER_HOUR = (60 * 60)/LARGE_DATA_COLLECTION_PERIOD;
+
+ /** Keeps track of th last reported latency value */
+ private AtomicLong lastLatency = new AtomicLong(0);
+
+ /**
+ * Queue of all latency values reported. The short term data collector clears this queue up
+ * time to time thus ensuring it doesn't grow indefinitely.
+ */
+ private Queue<Long> latencyDataQueue = new ConcurrentLinkedQueue<Long>();
+
+ /**
+ * Queue of samples collected by the short term data collector. This is maintained
+ * as a fixed length queue
+ */
+ private Queue<Long> shortTermLatencyDataQueue = new LinkedList<Long>();
+
+ /**
+ * Queue of samples collected by the long term data collector. This is maintained
+ * as a fixed length queue
+ */
+ private Queue<Long> longTermLatencyDataQueue = new LinkedList<Long>();
+
+ /** Scheduled executor on which data collectors are executed */
+ private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
+ private double allTimeAvgLatency = 0.0;
+ private int count = 0;
+ private Date resetTime = Calendar.getInstance().getTime();
+
+ private String name;
+
+ public LatencyView(boolean isHttps) throws AxisFault {
+ name = "nio-http" + (isHttps ? "s" : "");
+ scheduler.scheduleAtFixedRate(new ShortTermDataCollector(), SMALL_DATA_COLLECTION_PERIOD,
+ SMALL_DATA_COLLECTION_PERIOD, TimeUnit.SECONDS);
+ scheduler.scheduleAtFixedRate(new LongTermDataCollector(), LARGE_DATA_COLLECTION_PERIOD,
+ LARGE_DATA_COLLECTION_PERIOD, TimeUnit.SECONDS);
+
+ MBeanRegistrar.getInstance().registerMBean(this, NHTTP_LATENCY_VIEW, name);
+
+ }
+
+ public void destroy() {
+ MBeanRegistrar.getInstance().unRegisterMBean(NHTTP_LATENCY_VIEW, name);
+ scheduler.shutdownNow();
+ }
+
+ /**
+ * Report the timestamp values captured during mediating messages back and forth
+ *
+ * @param reqArrival The request arrival time
+ * @param reqDeparture The request departure time (backend connection establishment)
+ * @param resArrival The resoponse arrival time
+ * @param resDeparture The response departure time
+ */
+ public void notifyTimes(long reqArrival, long reqDeparture,
+ long resArrival, long resDeparture) {
+
+ long latency = (resDeparture - reqArrival) - (resArrival - reqDeparture);
+ lastLatency.set(latency);
+ latencyDataQueue.offer(latency);
+ }
+
+ public double getAllTimeAvgLatency() {
+ return allTimeAvgLatency;
+ }
+
+ public double getLastMinuteAvgLatency() {
+ return getAverageLatencyByMinute(1);
+ }
+
+ public double getLast5MinuteAvgLatency() {
+ return getAverageLatencyByMinute(5);
+ }
+
+ public double getLast15MinuteAvgLatency() {
+ return getAverageLatencyByMinute(15);
+ }
+
+ public double getLastHourAvgLatency() {
+ return getAverageLatencyByHour(1);
+ }
+
+ public double getLast8HourAvgLatency() {
+ return getAverageLatencyByHour(8);
+ }
+
+ public double getLast24HourAvgLatency() {
+ return getAverageLatencyByHour(24);
+ }
+
+ public void reset() {
+ lastLatency.set(0);
+ allTimeAvgLatency = 0.0;
+ latencyDataQueue.clear();
+ shortTermLatencyDataQueue.clear();
+ longTermLatencyDataQueue.clear();
+ count = 0;
+ resetTime = Calendar.getInstance().getTime();
+ }
+
+ public Date getLastResetTime() {
+ return resetTime;
+ }
+
+ private double getAverageLatencyByMinute(int n) {
+ int samples = n * SAMPLES_PER_MINUTE;
+ double sum = 0.0;
+ Long[] array = shortTermLatencyDataQueue.toArray(new Long[shortTermLatencyDataQueue.size()]);
+
+ if (samples > array.length) {
+ // If we don't have enough samples collected yet
+ // add up everything we have
+ samples = array.length;
+ for (Long anArray : array) {
+ sum += anArray;
+ }
+ } else {
+ // We have enough samples to make the right calculation
+ // Add up starting from the end of the queue (to give the most recent values)
+ for (int i = 0; i < samples; i++) {
+ sum += array[array.length - 1 - i];
+ }
+ }
+
+ if (samples == 0) {
+ return 0.0;
+ }
+ return sum/samples;
+ }
+
+ private double getAverageLatencyByHour(int n) {
+ int samples = n * SAMPLES_PER_HOUR;
+ double sum = 0.0;
+ Long[] array = longTermLatencyDataQueue.toArray(new Long[longTermLatencyDataQueue.size()]);
+
+ if (samples > array.length) {
+ samples = array.length;
+ for (Long anArray : array) {
+ sum += anArray;
+ }
+ } else {
+ for (int i = 0; i < samples; i++) {
+ sum += array[array.length - 1 - i];
+ }
+ }
+
+ if (samples == 0) {
+ return 0.0;
+ }
+ return sum/samples;
+ }
+
+ private class ShortTermDataCollector implements Runnable {
+ public void run() {
+ long latency = lastLatency.get();
+
+ // calculate all time average latency
+ int size = latencyDataQueue.size();
+ if (size > 0) {
+ long sum = 0;
+ for (int i = 0; i < size; i++) {
+ sum += latencyDataQueue.poll();
+ }
+ allTimeAvgLatency = (allTimeAvgLatency * count + sum)/(count + size);
+ count = count + size;
+ }
+
+ if (shortTermLatencyDataQueue.size() == 0 && latency == 0) {
+ // we haven't started collecting data yet - skip ahead...
+ return;
+ }
+
+ // take a sample for the short term latency calculation
+ if (shortTermLatencyDataQueue.size() == SAMPLES_PER_MINUTE * 15) {
+ shortTermLatencyDataQueue.remove();
+ }
+ shortTermLatencyDataQueue.offer(latency);
+ }
+ }
+
+ private class LongTermDataCollector implements Runnable {
+ public void run() {
+ long latency = lastLatency.get();
+ if (longTermLatencyDataQueue.size() == 0 && latency == 0) {
+ return;
+ }
+
+ if (longTermLatencyDataQueue.size() == SAMPLES_PER_HOUR * 24) {
+ longTermLatencyDataQueue.remove();
+ }
+ longTermLatencyDataQueue.offer(latency);
+ }
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyViewMBean.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyViewMBean.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyViewMBean.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyViewMBean.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,37 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import java.util.Date;
+
+public interface LatencyViewMBean {
+
+ public double getAllTimeAvgLatency();
+ public double getLastMinuteAvgLatency();
+ public double getLast5MinuteAvgLatency();
+ public double getLast15MinuteAvgLatency();
+ public double getLastHourAvgLatency();
+ public double getLast8HourAvgLatency();
+ public double getLast24HourAvgLatency();
+ public void reset();
+ public Date getLastResetTime();
+
+}
+
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/MBeanRegistrar.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/MBeanRegistrar.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/MBeanRegistrar.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/MBeanRegistrar.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import org.apache.axis2.AxisFault;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import java.lang.management.ManagementFactory;
+import java.util.Set;
+
+public class MBeanRegistrar {
+
+ private static final MBeanRegistrar ourInstance = new MBeanRegistrar();
+ private static final Log log = LogFactory.getLog(MBeanRegistrar.class);
+
+ public static MBeanRegistrar getInstance() {
+ return ourInstance;
+ }
+
+ private MBeanRegistrar() {
+ }
+
+ public void registerMBean(Object mBeanInstance, String category, String id) throws AxisFault {
+ assertNull(mBeanInstance, "MBean instance is null");
+ assertNull(category, "MBean instance category is null");
+ assertNull(id, "MBean instance name is null");
+ try {
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ ObjectName name = new ObjectName(getObjectName(category, id));
+ Set set = mbs.queryNames(name, null);
+ if (set != null && set.isEmpty()) {
+ mbs.registerMBean(mBeanInstance, name);
+ } else {
+ mbs.unregisterMBean(name);
+ mbs.registerMBean(mBeanInstance, name);
+ }
+ } catch (Exception e) {
+ log.warn("Error registering a MBean with name ' " + id +
+ " ' and category name ' " + category + "' for JMX management", e);
+ }
+ }
+
+ public void unRegisterMBean(String category, String id) {
+ try {
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ ObjectName objName = new ObjectName(getObjectName(category, id));
+ if (mbs.isRegistered(objName)) {
+ mbs.unregisterMBean(objName);
+ }
+ } catch (Exception e) {
+ log.warn("Error un-registering a MBean with name ' " + id +
+ " ' and category name ' " + category + "' for JMX management", e);
+ }
+ }
+
+ private String getObjectName(String category, String id) {
+
+ String jmxAgentName = System.getProperty("jmx.agent.name");
+ if (jmxAgentName == null || "".equals(jmxAgentName)) {
+ jmxAgentName = "org.apache.synapse";
+ }
+ return jmxAgentName + ":Type=" + category + ",Name=" + id;
+ }
+
+ private void assertNull(String name, String msg) throws AxisFault {
+ if (name == null || "".equals(name)) {
+ handleException(msg);
+ }
+ }
+
+ private void assertNull(Object object, String msg) throws AxisFault {
+ if (object == null) {
+ handleException(msg);
+ }
+ }
+
+ private static void handleException(String msg) throws AxisFault {
+ log.error(msg);
+ throw new AxisFault(msg);
+ }
+
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/PassThroughTransportMetricsCollector.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/PassThroughTransportMetricsCollector.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/PassThroughTransportMetricsCollector.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/PassThroughTransportMetricsCollector.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,76 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.transport.base.MetricsCollector;
+
+/**
+ * <p>This simple extension of the Axis2 transport MetricsCollector implementation,
+ * maintains a ConnectionsView instance, which is updated based on the events fired
+ * by the pass-through transport implementation. In addition to the usual events handled
+ * by the Axis2 MetricsCollector, this implementation handles two new events:</p>
+ * <ul>
+ * <li>connected (this should get called whenever a new connection is created)</li>
+ * <li>disconnected (this should get called whenever an existing connection is closed)</li>
+ * <ul>
+ * <p>These new events are used to update the ConnectionsView at runtime.</p>
+ */
+public class PassThroughTransportMetricsCollector extends MetricsCollector {
+
+ private ConnectionsView view;
+ private boolean listener;
+
+ public PassThroughTransportMetricsCollector(boolean listener, boolean isHttps)
+ throws AxisFault {
+ this.listener = listener;
+ String name = "http" + (isHttps ? "s" : "") + "-" + (listener ? "listener" : "sender");
+ this.view = new ConnectionsView(name);
+ }
+
+ public void destroy() {
+ view.destroy();
+ }
+
+ public void connected() {
+ view.connected();
+ }
+
+ public void disconnected() {
+ view.disconnected();
+ }
+
+ @Override
+ public void notifyReceivedMessageSize(long l) {
+ super.notifyReceivedMessageSize(l);
+ if (l > 0) {
+ view.notifyMessageSize(l, listener);
+ }
+ }
+
+ @Override
+ public void notifySentMessageSize(long l) {
+ super.notifySentMessageSize(l);
+ if (l > 0) {
+ view.notifyMessageSize(l, !listener);
+ }
+ }
+}
+
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportView.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportView.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportView.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportView.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,223 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import org.apache.axis2.AxisFault;
+import org.apache.synapse.transport.passthru.PassThroughHttpListener;
+import org.apache.synapse.transport.passthru.PassThroughHttpSender;
+
+import java.util.Map;
+import java.util.concurrent.ThreadPoolExecutor;
+
+public class TransportView implements TransportViewMBean {
+
+ private PassThroughHttpListener listener = null;
+
+ private PassThroughHttpSender sender = null;
+
+ private PassThroughTransportMetricsCollector metrics = null;
+
+ private ThreadPoolExecutor threadPool = null;
+
+ public TransportView(PassThroughHttpListener listener,
+ PassThroughHttpSender sender,
+ PassThroughTransportMetricsCollector metrics,
+ ThreadPoolExecutor threadPool) throws AxisFault {
+ this.listener = listener;
+ this.metrics = metrics;
+ this.threadPool = threadPool;
+ this.sender = sender;
+ }
+
+ public void pause() throws AxisFault {
+ if (listener != null) {
+ listener.pause();
+ } else if (sender != null) {
+ sender.pause();
+ }
+ }
+
+ public void resume() throws AxisFault {
+ if (listener != null) {
+ listener.resume();
+ } else if (sender != null) {
+ sender.resume();
+ }
+ }
+
+ public void maintenenceShutdown(long l) throws AxisFault {
+ if (listener != null) {
+ listener.maintenanceShutdown(l);
+ } else if (sender != null) {
+ sender.maintenanceShutdown(l);
+ }
+ }
+
+ public int getActiveThreadCount() {
+ if (threadPool != null) {
+ return threadPool.getActiveCount();
+ }
+ return 0;
+ }
+
+ public int getQueueSize() {
+ if (threadPool != null && threadPool.getQueue() != null) {
+ return threadPool.getQueue().size();
+ }
+ return 0;
+ }
+
+ public long getMessagesReceived() {
+ if (metrics != null) {
+ return metrics.getMessagesReceived();
+ }
+ return -1;
+ }
+
+ public long getFaultsReceiving() {
+ if (metrics != null) {
+ return metrics.getFaultsReceiving();
+ }
+ return -1;
+ }
+
+ public long getBytesReceived() {
+ if (metrics != null) {
+ return metrics.getBytesReceived();
+ }
+ return -1;
+ }
+
+ public long getMessagesSent() {
+ if (metrics != null) {
+ return metrics.getMessagesSent();
+ }
+ return -1;
+ }
+
+ public long getFaultsSending() {
+ if (metrics != null) {
+ return metrics.getFaultsSending();
+ }
+ return -1;
+ }
+
+ public long getBytesSent() {
+ if (metrics != null) {
+ return metrics.getBytesSent();
+ }
+ return -1;
+ }
+
+ public long getTimeoutsReceiving() {
+ if (metrics != null) {
+ return metrics.getTimeoutsReceiving();
+ }
+ return -1;
+ }
+
+ public long getTimeoutsSending() {
+ if (metrics != null) {
+ return metrics.getTimeoutsSending();
+ }
+ return -1;
+ }
+
+ public long getMinSizeReceived() {
+ if (metrics != null) {
+ return metrics.getMinSizeReceived();
+ }
+ return -1;
+ }
+
+ public long getMaxSizeReceived() {
+ if (metrics != null) {
+ return metrics.getMaxSizeReceived();
+ }
+ return -1;
+ }
+
+ public double getAvgSizeReceived() {
+ if (metrics != null) {
+ return metrics.getAvgSizeReceived();
+ }
+ return -1;
+ }
+
+ public long getMinSizeSent() {
+ if (metrics != null) {
+ return metrics.getMinSizeSent();
+ }
+ return -1;
+ }
+
+ public long getMaxSizeSent() {
+ if (metrics != null) {
+ return metrics.getMaxSizeSent();
+ }
+ return -1;
+ }
+
+ public double getAvgSizeSent() {
+ if (metrics != null) {
+ return metrics.getAvgSizeSent();
+ }
+ return -1;
+ }
+
+ public Map getResponseCodeTable() {
+ if (metrics != null) {
+ return metrics.getResponseCodeTable();
+ }
+ return null;
+ }
+
+ public void start() throws Exception {
+ if (listener != null) {
+ listener.start();
+ }
+ }
+
+ public void stop() throws Exception {
+ if (listener != null) {
+ listener.stop();
+ }
+ }
+
+ public void resetStatistics() {
+ if (metrics != null) {
+ metrics.reset();
+ }
+ }
+
+ public long getLastResetTime() {
+ if (metrics != null) {
+ return metrics.getLastResetTime();
+ }
+ return -1;
+ }
+
+ public long getMetricsWindow() {
+ if (metrics != null) {
+ return System.currentTimeMillis() - metrics.getLastResetTime();
+ }
+ return -1;
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportViewMBean.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportViewMBean.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportViewMBean.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/TransportViewMBean.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.synapse.transport.passthru.jmx;
+
+import java.util.Map;
+
+public interface TransportViewMBean {
+
+ // JMX Attributes
+ public long getMessagesReceived();
+ public long getFaultsReceiving();
+ public long getTimeoutsReceiving();
+ public long getMessagesSent();
+ public long getFaultsSending();
+ public long getTimeoutsSending();
+ public long getBytesReceived();
+ public long getBytesSent();
+ public long getMinSizeReceived();
+ public long getMaxSizeReceived();
+ public double getAvgSizeReceived();
+ public long getMinSizeSent();
+ public long getMaxSizeSent();
+ public double getAvgSizeSent();
+ public int getActiveThreadCount();
+ public int getQueueSize();
+ public Map getResponseCodeTable();
+
+ // JMX Operations
+ public void start() throws Exception;
+ public void stop() throws Exception;
+ public void pause() throws Exception;
+ public void resume() throws Exception;
+ public void maintenenceShutdown(long seconds) throws Exception;
+
+ public void resetStatistics();
+ public long getLastResetTime();
+ public long getMetricsWindow();
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingIOSession.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingIOSession.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingIOSession.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingIOSession.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,225 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.logging;
+
+import org.apache.http.nio.reactor.IOSession;
+import org.apache.http.nio.reactor.SessionBufferStatus;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.util.concurrent.atomic.AtomicLong;
+import java.nio.channels.ByteChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.ByteBuffer;
+import java.net.SocketAddress;
+import java.io.IOException;
+
+class LoggingIOSession implements IOSession {
+
+ private static AtomicLong COUNT = new AtomicLong(0);
+
+ private final Log log;
+ private final Wire wirelog;
+ private final IOSession session;
+ private final ByteChannel channel;
+ private final String id;
+
+ public LoggingIOSession(
+ final Log wirelog,
+ final IOSession session,
+ final String id) {
+ super();
+ if (session == null) {
+ throw new IllegalArgumentException("I/O session may not be null");
+ }
+ this.session = session;
+ this.channel = new LoggingByteChannel();
+ this.id = id + "-" + COUNT.incrementAndGet();
+ this.log = LogFactory.getLog(session.getClass());
+ this.wirelog = new Wire(wirelog);
+ }
+
+ public int getStatus() {
+ return this.session.getStatus();
+ }
+
+ public ByteChannel channel() {
+ return this.channel;
+ }
+
+ public SocketAddress getLocalAddress() {
+ return this.session.getLocalAddress();
+ }
+
+ public SocketAddress getRemoteAddress() {
+ return this.session.getRemoteAddress();
+ }
+
+ public int getEventMask() {
+ return this.session.getEventMask();
+ }
+
+ private static String formatOps(int ops) {
+ StringBuffer buffer = new StringBuffer(6);
+ buffer.append('[');
+ if ((ops & SelectionKey.OP_READ) > 0) {
+ buffer.append('r');
+ }
+ if ((ops & SelectionKey.OP_WRITE) > 0) {
+ buffer.append('w');
+ }
+ if ((ops & SelectionKey.OP_ACCEPT) > 0) {
+ buffer.append('a');
+ }
+ if ((ops & SelectionKey.OP_CONNECT) > 0) {
+ buffer.append('c');
+ }
+ buffer.append(']');
+ return buffer.toString();
+ }
+
+ public void setEventMask(int ops) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Set event mask "
+ + formatOps(ops));
+ }
+ this.session.setEventMask(ops);
+ }
+
+ public void setEvent(int op) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Set event "
+ + formatOps(op));
+ }
+ this.session.setEvent(op);
+ }
+
+ public void clearEvent(int op) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Clear event "
+ + formatOps(op));
+ }
+ this.session.clearEvent(op);
+ }
+
+ public void close() {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Close");
+ }
+ this.session.close();
+ }
+
+ public boolean isClosed() {
+ return this.session.isClosed();
+ }
+
+ public void shutdown() {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Shutdown");
+ }
+ this.session.shutdown();
+ }
+
+ public int getSocketTimeout() {
+ return this.session.getSocketTimeout();
+ }
+
+ public void setSocketTimeout(int timeout) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Set timeout "
+ + timeout);
+ }
+ this.session.setSocketTimeout(timeout);
+ }
+
+ public void setBufferStatus(final SessionBufferStatus status) {
+ this.session.setBufferStatus(status);
+ }
+
+ public boolean hasBufferedInput() {
+ return this.session.hasBufferedInput();
+ }
+
+ public boolean hasBufferedOutput() {
+ return this.session.hasBufferedOutput();
+ }
+
+ public Object getAttribute(final String name) {
+ return this.session.getAttribute(name);
+ }
+
+ public void setAttribute(final String name, final Object obj) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Set attribute "
+ + name);
+ }
+ this.session.setAttribute(name, obj);
+ }
+
+ public Object removeAttribute(final String name) {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("I/O session " + this.id + " " + this.session + ": Remove attribute "
+ + name);
+ }
+ return this.session.removeAttribute(name);
+ }
+
+ class LoggingByteChannel implements ByteChannel {
+
+ public int read(final ByteBuffer dst) throws IOException {
+ int bytesRead = session.channel().read(dst);
+ if (log.isDebugEnabled()) {
+ log.debug("I/O session " + id + " " + session + ": " + bytesRead + " bytes read");
+ }
+ if (bytesRead > 0 && wirelog.isEnabled()) {
+ ByteBuffer b = dst.duplicate();
+ int p = b.position();
+ b.limit(p);
+ b.position(p - bytesRead);
+ wirelog.input(b);
+ }
+ return bytesRead;
+ }
+
+ public int write(final ByteBuffer src) throws IOException {
+ int byteWritten = session.channel().write(src);
+ if (log.isDebugEnabled()) {
+ log.debug("I/O session " + id + " " + session + ": " + byteWritten + " bytes written");
+ }
+ if (byteWritten > 0 && wirelog.isEnabled()) {
+ ByteBuffer b = src.duplicate();
+ int p = b.position();
+ b.limit(p);
+ b.position(p - byteWritten);
+ wirelog.output(b);
+ }
+ return byteWritten;
+ }
+
+ public void close() throws IOException {
+ if (log.isDebugEnabled()) {
+ log.debug("I/O session " + id + " " + session + ": Channel close");
+ }
+ session.channel().close();
+ }
+
+ public boolean isOpen() {
+ return session.channel().isOpen();
+ }
+
+ }
+}
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpSourceConnection.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpSourceConnection.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpSourceConnection.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpSourceConnection.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,157 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.logging;
+
+import org.apache.http.impl.nio.DefaultNHttpServerConnection;
+import org.apache.http.nio.reactor.IOSession;
+import org.apache.http.nio.reactor.SessionOutputBuffer;
+import org.apache.http.nio.reactor.SessionInputBuffer;
+import org.apache.http.nio.util.ByteBufferAllocator;
+import org.apache.http.nio.NHttpServiceHandler;
+import org.apache.http.nio.NHttpMessageWriter;
+import org.apache.http.nio.NHttpMessageParser;
+import org.apache.http.*;
+import org.apache.http.params.HttpParams;
+import org.apache.commons.logging.Log;
+
+import java.io.IOException;
+import java.nio.channels.ReadableByteChannel;
+
+public class LoggingNHttpSourceConnection extends DefaultNHttpServerConnection {
+ private final Log log;
+ private final Log headerLog;
+
+ public LoggingNHttpSourceConnection(
+ final Log log,
+ final Log headerlog,
+ final IOSession session,
+ final HttpRequestFactory requestFactory,
+ final ByteBufferAllocator allocator,
+ final HttpParams params) {
+ super(session, requestFactory, allocator, params);
+ this.log = log;
+ this.headerLog = headerlog;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.log.debug("Close connection");
+ super.close();
+ }
+
+ @Override
+ public void shutdown() throws IOException {
+ this.log.debug("Shutdown connection");
+ super.shutdown();
+ }
+
+ @Override
+ public void submitResponse(final HttpResponse response) throws IOException, HttpException {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("HTTP connection " + this + ": " + response.getStatusLine().toString());
+ }
+ super.submitResponse(response);
+ }
+
+ @Override
+ public void consumeInput(final NHttpServiceHandler handler) {
+ this.log.debug("Consume input");
+ super.consumeInput(handler);
+ }
+
+ @Override
+ public void produceOutput(final NHttpServiceHandler handler) {
+ this.log.debug("Produce output");
+ super.produceOutput(handler);
+ }
+
+ @Override
+ protected NHttpMessageWriter createResponseWriter(
+ final SessionOutputBuffer buffer,
+ final HttpParams params) {
+ return new LoggingNHttpMessageWriter(
+ super.createResponseWriter(buffer, params));
+ }
+
+ @Override
+ protected NHttpMessageParser createRequestParser(
+ final SessionInputBuffer buffer,
+ final HttpRequestFactory requestFactory,
+ final HttpParams params) {
+ return new LoggingNHttpMessageParser(
+ super.createRequestParser(buffer, requestFactory, params));
+ }
+
+ class LoggingNHttpMessageWriter implements NHttpMessageWriter {
+
+ private final NHttpMessageWriter writer;
+
+ public LoggingNHttpMessageWriter(final NHttpMessageWriter writer) {
+ super();
+ this.writer = writer;
+ }
+
+ public void reset() {
+ this.writer.reset();
+ }
+
+ public void write(final HttpMessage message) throws IOException, HttpException {
+ if (message != null && headerLog.isDebugEnabled()) {
+ HttpResponse response = (HttpResponse) message;
+ headerLog.debug("<< " + response.getStatusLine().toString());
+ Header[] headers = response.getAllHeaders();
+ for (Header header : headers) {
+ headerLog.debug("<< " + header.toString());
+ }
+ }
+ this.writer.write(message);
+ }
+
+ }
+
+ class LoggingNHttpMessageParser implements NHttpMessageParser {
+
+ private final NHttpMessageParser parser;
+
+ public LoggingNHttpMessageParser(final NHttpMessageParser parser) {
+ super();
+ this.parser = parser;
+ }
+
+ public void reset() {
+ this.parser.reset();
+ }
+
+ public int fillBuffer(final ReadableByteChannel channel) throws IOException {
+ return this.parser.fillBuffer(channel);
+ }
+
+ public HttpMessage parse() throws IOException, HttpException {
+ HttpMessage message = this.parser.parse();
+ if (message != null && headerLog.isDebugEnabled()) {
+ HttpRequest request = (HttpRequest) message;
+ headerLog.debug(">> " + request.getRequestLine().toString());
+ Header[] headers = request.getAllHeaders();
+ for (Header header : headers) {
+ headerLog.debug(">> " + header.toString());
+ }
+ }
+ return message;
+ }
+ }
+}
+
Added: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpTargetConnection.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpTargetConnection.java?rev=1502386&view=auto
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpTargetConnection.java (added)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/logging/LoggingNHttpTargetConnection.java Thu Jul 11 22:18:46 2013
@@ -0,0 +1,157 @@
+/*
+ * Copyright 2013 Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.synapse.transport.passthru.logging;
+
+import org.apache.commons.logging.Log;
+import org.apache.http.nio.reactor.IOSession;
+import org.apache.http.nio.reactor.SessionOutputBuffer;
+import org.apache.http.nio.reactor.SessionInputBuffer;
+import org.apache.http.nio.util.ByteBufferAllocator;
+import org.apache.http.nio.NHttpClientHandler;
+import org.apache.http.nio.NHttpMessageWriter;
+import org.apache.http.nio.NHttpMessageParser;
+import org.apache.http.*;
+import org.apache.http.impl.nio.DefaultNHttpClientConnection;
+import org.apache.http.params.HttpParams;
+
+import java.io.IOException;
+import java.nio.channels.ReadableByteChannel;
+
+public class LoggingNHttpTargetConnection extends DefaultNHttpClientConnection {
+ private final Log log;
+ private final Log headerLog;
+
+ public LoggingNHttpTargetConnection(
+ final Log log,
+ final Log headerlog,
+ final IOSession iosession,
+ final HttpResponseFactory responseFactory,
+ final ByteBufferAllocator allocator,
+ final HttpParams params) {
+ super(iosession, responseFactory, allocator, params);
+ this.log = log;
+ this.headerLog = headerlog;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.log.debug("Close connection");
+ super.close();
+ }
+
+ @Override
+ public void shutdown() throws IOException {
+ this.log.debug("Shutdown connection");
+ super.shutdown();
+ }
+
+ @Override
+ public void submitRequest(final HttpRequest request) throws IOException, HttpException {
+ if (this.log.isDebugEnabled()) {
+ this.log.debug("HTTP connection " + this + ": " + request.getRequestLine().toString());
+ }
+ super.submitRequest(request);
+ }
+
+ @Override
+ public void consumeInput(final NHttpClientHandler handler) {
+ this.log.debug("Consume input");
+ super.consumeInput(handler);
+ }
+
+ @Override
+ public void produceOutput(final NHttpClientHandler handler) {
+ this.log.debug("Produce output");
+ super.produceOutput(handler);
+ }
+
+ @Override
+ protected NHttpMessageWriter createRequestWriter(
+ final SessionOutputBuffer buffer,
+ final HttpParams params) {
+ return new LoggingNHttpMessageWriter(
+ super.createRequestWriter(buffer, params));
+ }
+
+ @Override
+ protected NHttpMessageParser createResponseParser(
+ final SessionInputBuffer buffer,
+ final HttpResponseFactory responseFactory,
+ final HttpParams params) {
+ return new LoggingNHttpMessageParser(
+ super.createResponseParser(buffer, responseFactory, params));
+ }
+
+ class LoggingNHttpMessageWriter implements NHttpMessageWriter {
+
+ private final NHttpMessageWriter writer;
+
+ public LoggingNHttpMessageWriter(final NHttpMessageWriter writer) {
+ super();
+ this.writer = writer;
+ }
+
+ public void reset() {
+ this.writer.reset();
+ }
+
+ public void write(final HttpMessage message) throws IOException, HttpException {
+ if (message != null && headerLog.isDebugEnabled()) {
+ HttpRequest request = (HttpRequest) message;
+ headerLog.debug(">> " + request.getRequestLine().toString());
+ Header[] headers = request.getAllHeaders();
+ for (Header header : headers) {
+ headerLog.debug(">> " + header.toString());
+ }
+ }
+ this.writer.write(message);
+ }
+
+ }
+
+ class LoggingNHttpMessageParser implements NHttpMessageParser {
+
+ private final NHttpMessageParser parser;
+
+ public LoggingNHttpMessageParser(final NHttpMessageParser parser) {
+ super();
+ this.parser = parser;
+ }
+
+ public void reset() {
+ this.parser.reset();
+ }
+
+ public int fillBuffer(final ReadableByteChannel channel) throws IOException {
+ return this.parser.fillBuffer(channel);
+ }
+
+ public HttpMessage parse() throws IOException, HttpException {
+ HttpMessage message = this.parser.parse();
+ if (message != null && headerLog.isDebugEnabled()) {
+ HttpResponse response = (HttpResponse) message;
+ headerLog.debug("<< " + response.getStatusLine().toString());
+ Header[] headers = response.getAllHeaders();
+ for (Header header : headers) {
+ headerLog.debug("<< " + header.toString());
+ }
+ }
+ return message;
+ }
+
+ }
+}