Apache
/
jclouds
mirror of https://github.com/apache/jclouds.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

issue 66: refactored http clients to not be bound to a single endpoint such that redirects can be assigned to another host 

git-svn-id: http://jclouds.googlecode.com/svn/trunk@1458 3d8758e0-26b5-11de-8745-db77d3ebf521
This commit is contained in:
adrian.f.cole 2009-06-24 18:27:39 +00:00
parent 00cc4c4c4e
commit 78d0b7afd3
10 changed files with 427 additions and 632 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/config/HttpNioConnectionPoolClientModule.java
View File

@ -23,57 +23,131 @@
*/ */
package org.jclouds.http.httpnio.config; package org.jclouds.http.httpnio.config;
import java.net.InetSocketAddress; import java.util.concurrent.ArrayBlockingQueue;
import java.net.MalformedURLException; import java.util.concurrent.BlockingQueue;
import java.net.URI; import java.util.concurrent.LinkedBlockingQueue;
import org.jclouds.http.HttpConstants; import org.apache.http.ConnectionReuseStrategy;
import org.apache.http.HttpEntity;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.entity.BufferingNHttpEntity;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
import org.apache.http.nio.protocol.NHttpRequestExecutionHandler;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.nio.util.ByteBufferAllocator;
import org.apache.http.nio.util.HeapByteBufferAllocator;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.CoreProtocolPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;
import org.jclouds.command.pool.PoolConstants;
import org.jclouds.command.pool.config.FutureCommandConnectionPoolClientModule;
import org.jclouds.http.HttpFutureCommand;
import org.jclouds.http.HttpFutureCommandClient; import org.jclouds.http.HttpFutureCommandClient;
import org.jclouds.http.config.HttpFutureCommandClientModule;
import org.jclouds.http.httpnio.config.internal.NonSSLHttpNioConnectionPoolClientModule;
import org.jclouds.http.httpnio.config.internal.SSLHttpNioConnectionPoolClientModule;
import org.jclouds.http.httpnio.pool.HttpNioConnectionPoolClient; import org.jclouds.http.httpnio.pool.HttpNioConnectionPoolClient;
import org.jclouds.http.httpnio.pool.HttpNioFutureCommandConnectionPool;
import org.jclouds.http.httpnio.pool.HttpNioFutureCommandExecutionHandler;
import com.google.inject.AbstractModule; import com.google.inject.Inject;
import com.google.inject.Provides; import com.google.inject.Provides;
import com.google.inject.Scopes;
import com.google.inject.Singleton; import com.google.inject.Singleton;
import com.google.inject.TypeLiteral;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.FactoryProvider;
import com.google.inject.name.Named; import com.google.inject.name.Named;
/** /**
* Configures {@link HttpNioConnectionPoolClient}
* *
* @author Adrian Cole * @author Adrian Cole
*/ */
@HttpFutureCommandClientModule public class HttpNioConnectionPoolClientModule extends
public class HttpNioConnectionPoolClientModule extends AbstractModule { FutureCommandConnectionPoolClientModule<NHttpConnection> {
@Named(HttpConstants.PROPERTY_HTTP_SECURE) @Provides
boolean isSecure; // @Singleton per uri...
public AsyncNHttpClientHandler provideAsyncNttpClientHandler(BasicHttpProcessor httpProcessor,
NHttpRequestExecutionHandler execHandler, ConnectionReuseStrategy connStrategy,
ByteBufferAllocator allocator, HttpParams params) {
return new AsyncNHttpClientHandler(httpProcessor, execHandler, connStrategy, allocator,
params);
}
@Provides
@Singleton
public BasicHttpProcessor provideClientProcessor() {
BasicHttpProcessor httpproc = new BasicHttpProcessor();
httpproc.addInterceptor(new RequestContent());
httpproc.addInterceptor(new RequestTargetHost());
httpproc.addInterceptor(new RequestConnControl());
httpproc.addInterceptor(new RequestUserAgent());
httpproc.addInterceptor(new RequestExpectContinue());
return httpproc;
}
@Provides
@Singleton
public HttpParams provideHttpParams() {
HttpParams params = new BasicHttpParams();
params.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 5000).setIntParameter(
CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024).setBooleanParameter(
CoreConnectionPNames.STALE_CONNECTION_CHECK, false).setBooleanParameter(
CoreConnectionPNames.TCP_NODELAY, true).setParameter(
CoreProtocolPNames.ORIGIN_SERVER, "jclouds/1.0");
return params;
}
protected void configure() {
super.configure();
bind(HttpFutureCommandClient.class).to(HttpNioConnectionPoolClient.class);
bind(new TypeLiteral<BlockingQueue<HttpFutureCommand<?>>>() {
}).to(new TypeLiteral<LinkedBlockingQueue<HttpFutureCommand<?>>>() {
}).in(Scopes.SINGLETON);
bind(HttpNioFutureCommandExecutionHandler.ConsumingNHttpEntityFactory.class).toProvider(
FactoryProvider.newFactory(
HttpNioFutureCommandExecutionHandler.ConsumingNHttpEntityFactory.class,
InjectableBufferingNHttpEntity.class));// .in(Scopes.SINGLETON); but per URI
bind(NHttpRequestExecutionHandler.class).to(HttpNioFutureCommandExecutionHandler.class).in(
Scopes.SINGLETON);
bind(ConnectionReuseStrategy.class).to(DefaultConnectionReuseStrategy.class).in(
Scopes.SINGLETON);
bind(ByteBufferAllocator.class).to(HeapByteBufferAllocator.class);
bind(HttpNioFutureCommandConnectionPool.Factory.class).toProvider(
FactoryProvider.newFactory(
new TypeLiteral<HttpNioFutureCommandConnectionPool.Factory>() {
}, new TypeLiteral<HttpNioFutureCommandConnectionPool>() {
}));
}
static class InjectableBufferingNHttpEntity extends BufferingNHttpEntity {
@Inject
public InjectableBufferingNHttpEntity(@Assisted HttpEntity httpEntity,
ByteBufferAllocator allocator) {
super(httpEntity, allocator);
}
}
@Override @Override
protected void configure() { public BlockingQueue<NHttpConnection> provideAvailablePool(
requestInjection(this); @Named(PoolConstants.PROPERTY_POOL_MAX_CONNECTIONS) int max) throws Exception {
if (isSecure) return new ArrayBlockingQueue<NHttpConnection>(max, true);
install(new SSLHttpNioConnectionPoolClientModule());
else
install(new NonSSLHttpNioConnectionPoolClientModule());
bind(HttpFutureCommandClient.class).to(HttpNioConnectionPoolClient.class);
} }
@Singleton
@Provides @Provides
protected InetSocketAddress provideAddress(URI endPoint) { // @Singleton per uri...
return new InetSocketAddress(endPoint.getHost(), endPoint.getPort()); public DefaultConnectingIOReactor provideDefaultConnectingIOReactor(
@Named(PoolConstants.PROPERTY_POOL_IO_WORKER_THREADS) int ioWorkerThreads,
HttpParams params) throws IOReactorException {
return new DefaultConnectingIOReactor(ioWorkerThreads, params);
} }
@Singleton
@Provides
protected URI provideAddress(@Named(HttpConstants.PROPERTY_HTTP_ADDRESS) String address,
@Named(HttpConstants.PROPERTY_HTTP_PORT) int port,
@Named(HttpConstants.PROPERTY_HTTP_SECURE) boolean isSecure)
throws MalformedURLException {
return URI.create(String.format("%1$s://%2$s:%3$s", isSecure ? "https" : "http", address,
port));
}
} }

172
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/config/internal/BaseHttpNioConnectionPoolClientModule.java
View File

@ -1,172 +0,0 @@
/**
*
* Copyright (C) 2009 Global Cloud Specialists, Inc. <info@globalcloudspecialists.com>
*
* ====================================================================
* 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.jclouds.http.httpnio.config.internal;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.http.ConnectionReuseStrategy;
import org.apache.http.HttpEntity;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.entity.BufferingNHttpEntity;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
import org.apache.http.nio.protocol.NHttpRequestExecutionHandler;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.nio.util.ByteBufferAllocator;
import org.apache.http.nio.util.HeapByteBufferAllocator;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.CoreProtocolPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;
import org.jclouds.command.pool.PoolConstants;
import org.jclouds.command.pool.config.FutureCommandConnectionPoolClientModule;
import org.jclouds.http.HttpFutureCommand;
import org.jclouds.http.httpnio.pool.HttpNioFutureCommandConnectionHandle;
import org.jclouds.http.httpnio.pool.HttpNioFutureCommandConnectionPool;
import org.jclouds.http.httpnio.pool.HttpNioFutureCommandExecutionHandler;
import com.google.inject.Inject;
import com.google.inject.Provides;
import com.google.inject.Scopes;
import com.google.inject.Singleton;
import com.google.inject.TypeLiteral;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.FactoryProvider;
import com.google.inject.name.Named;
/**
* // TODO: Adrian: Document this!
*
* @author Adrian Cole
*/
public abstract class BaseHttpNioConnectionPoolClientModule extends
FutureCommandConnectionPoolClientModule<NHttpConnection> {
@Provides
@Singleton
public AsyncNHttpClientHandler provideAsyncNttpClientHandler(
BasicHttpProcessor httpProcessor,
NHttpRequestExecutionHandler execHandler,
ConnectionReuseStrategy connStrategy,
ByteBufferAllocator allocator, HttpParams params) {
return new AsyncNHttpClientHandler(httpProcessor, execHandler,
connStrategy, allocator, params);
}
@Provides
@Singleton
public BasicHttpProcessor provideClientProcessor() {
BasicHttpProcessor httpproc = new BasicHttpProcessor();
httpproc.addInterceptor(new RequestContent());
httpproc.addInterceptor(new RequestTargetHost());
httpproc.addInterceptor(new RequestConnControl());
httpproc.addInterceptor(new RequestUserAgent());
httpproc.addInterceptor(new RequestExpectContinue());
return httpproc;
}
@Provides
@Singleton
public HttpParams provideHttpParams() {
HttpParams params = new BasicHttpParams();
params.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 5000)
.setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE,
8 * 1024).setBooleanParameter(
CoreConnectionPNames.STALE_CONNECTION_CHECK, false)
.setBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true)
.setParameter(CoreProtocolPNames.ORIGIN_SERVER, "jclouds/1.0");
return params;
}
protected void configure() {
super.configure();
bind(new TypeLiteral<BlockingQueue<HttpFutureCommand<?>>>() {
}).to(new TypeLiteral<LinkedBlockingQueue<HttpFutureCommand<?>>>() {
}).in(Scopes.SINGLETON);
bind(
HttpNioFutureCommandExecutionHandler.ConsumingNHttpEntityFactory.class)
.toProvider(
FactoryProvider
.newFactory(
HttpNioFutureCommandExecutionHandler.ConsumingNHttpEntityFactory.class,
InjectableBufferingNHttpEntity.class))
.in(Scopes.SINGLETON);
bind(NHttpRequestExecutionHandler.class).to(
HttpNioFutureCommandExecutionHandler.class)
.in(Scopes.SINGLETON);
bind(ConnectionReuseStrategy.class).to(
DefaultConnectionReuseStrategy.class).in(Scopes.SINGLETON);
bind(ByteBufferAllocator.class).to(HeapByteBufferAllocator.class);
bind(
HttpNioFutureCommandConnectionPool.FutureCommandConnectionHandleFactory.class)
.toProvider(
FactoryProvider
.newFactory(
new TypeLiteral<HttpNioFutureCommandConnectionPool.FutureCommandConnectionHandleFactory>() {
},
new TypeLiteral<HttpNioFutureCommandConnectionHandle>() {
}));
}
static class InjectableBufferingNHttpEntity extends BufferingNHttpEntity {
@Inject
public InjectableBufferingNHttpEntity(@Assisted HttpEntity httpEntity,
ByteBufferAllocator allocator) {
super(httpEntity, allocator);
}
}
@Override
public BlockingQueue<NHttpConnection> provideAvailablePool(
@Named(PoolConstants.PROPERTY_POOL_MAX_CONNECTIONS) int max)
throws Exception {
return new ArrayBlockingQueue<NHttpConnection>(max, true);
}
@Provides
@Singleton
public abstract IOEventDispatch provideClientEventDispatch(
AsyncNHttpClientHandler handler, HttpParams params)
throws Exception;
@Provides
@Singleton
public DefaultConnectingIOReactor provideDefaultConnectingIOReactor(
@Named(PoolConstants.PROPERTY_POOL_IO_WORKER_THREADS) int ioWorkerThreads,
HttpParams params) throws IOReactorException {
return new DefaultConnectingIOReactor(ioWorkerThreads, params);
}
}

43
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/config/internal/NonSSLHttpNioConnectionPoolClientModule.java
View File

@ -1,43 +0,0 @@
/**
*
* Copyright (C) 2009 Global Cloud Specialists, Inc. <info@globalcloudspecialists.com>
*
* ====================================================================
* 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.jclouds.http.httpnio.config.internal;
import org.apache.http.impl.nio.DefaultClientIOEventDispatch;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.params.HttpParams;
import org.jclouds.http.httpnio.config.internal.BaseHttpNioConnectionPoolClientModule;
/**
* // TODO: Adrian: Document this!
*
* @author Adrian Cole
*/
public class NonSSLHttpNioConnectionPoolClientModule extends BaseHttpNioConnectionPoolClientModule {
public IOEventDispatch provideClientEventDispatch(AsyncNHttpClientHandler handler, HttpParams params) throws Exception {
return new DefaultClientIOEventDispatch(
handler,
params);
}
}

56
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/config/internal/SSLHttpNioConnectionPoolClientModule.java
View File

@ -1,56 +0,0 @@
/**
*
* Copyright (C) 2009 Global Cloud Specialists, Inc. <info@globalcloudspecialists.com>
*
* ====================================================================
* 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.jclouds.http.httpnio.config.internal;
import org.apache.http.impl.nio.SSLClientIOEventDispatch;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
import org.apache.http.params.HttpParams;
import org.jclouds.http.httpnio.config.internal.BaseHttpNioConnectionPoolClientModule;
import javax.net.ssl.SSLContext;
/**
* // TODO: Adrian: Document this!
*
* @author Adrian Cole
*/
public class SSLHttpNioConnectionPoolClientModule extends BaseHttpNioConnectionPoolClientModule {
protected void configure() {
super.configure();
}
// note until a bug is fixed, you cannot annotate overriding methods with google annotations
// http://code.google.com/p/google-guice/issues/detail?id=347
@Override
public IOEventDispatch provideClientEventDispatch(AsyncNHttpClientHandler handler, HttpParams params) throws Exception {
SSLContext context = SSLContext.getInstance("TLS");
context.init(null, null, null);
return new SSLClientIOEventDispatch(
handler,
context,
params);
}
}

36
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/pool/HttpNioConnectionPoolClient.java
View File

@ -23,27 +23,33 @@
*/ */
package org.jclouds.http.httpnio.pool; package org.jclouds.http.httpnio.pool;
import com.google.inject.Inject; import java.net.URI;
import com.google.inject.Singleton;
import org.apache.http.nio.NHttpConnection;
import org.jclouds.command.pool.FutureCommandConnectionPoolClient;
import org.jclouds.http.*;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import org.apache.http.nio.NHttpConnection;
import org.jclouds.command.pool.FutureCommandConnectionPoolClient;
import org.jclouds.http.HttpException;
import org.jclouds.http.HttpFutureCommand;
import org.jclouds.http.HttpFutureCommandClient;
import org.jclouds.http.HttpRequest;
import org.jclouds.http.HttpRequestFilter;
import com.google.inject.Inject;
import com.google.inject.Singleton;
/** /**
* // TODO: Adrian: Document this! * // TODO: Adrian: Document this!
* *
* @author Adrian Cole * @author Adrian Cole
*/ */
@Singleton @Singleton
public class HttpNioConnectionPoolClient public class HttpNioConnectionPoolClient extends
extends FutureCommandConnectionPoolClient<URI, NHttpConnection, HttpFutureCommand<?>> implements
FutureCommandConnectionPoolClient<NHttpConnection, HttpFutureCommand<?>> HttpFutureCommandClient {
implements HttpFutureCommandClient {
private List<HttpRequestFilter> requestFilters = Collections.emptyList(); private List<HttpRequestFilter> requestFilters = Collections.emptyList();
public List<HttpRequestFilter> getRequestFilters() { public List<HttpRequestFilter> getRequestFilters() {
@ -69,13 +75,9 @@ public class HttpNioConnectionPoolClient
} }
@Inject @Inject
public HttpNioConnectionPoolClient( public HttpNioConnectionPoolClient(ExecutorService executor,
ExecutorService executor, HttpNioFutureCommandConnectionPool.Factory poolFactory,
HttpNioFutureCommandConnectionPool httpFutureCommandConnectionHandleNHttpConnectionNioFutureCommandConnectionPool,
BlockingQueue<HttpFutureCommand<?>> commandQueue) { BlockingQueue<HttpFutureCommand<?>> commandQueue) {
super( super(executor, poolFactory, commandQueue);
executor,
httpFutureCommandConnectionHandleNHttpConnectionNioFutureCommandConnectionPool,
commandQueue);
} }
} }

17
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/pool/HttpNioFutureCommandConnectionHandle.java
View File

@ -24,6 +24,7 @@
package org.jclouds.http.httpnio.pool; package org.jclouds.http.httpnio.pool;
import java.io.IOException; import java.io.IOException;
import java.net.URI;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore; import java.util.concurrent.Semaphore;
@ -31,7 +32,6 @@ import org.apache.http.nio.NHttpConnection;
import org.jclouds.command.pool.FutureCommandConnectionHandle; import org.jclouds.command.pool.FutureCommandConnectionHandle;
import org.jclouds.http.HttpFutureCommand; import org.jclouds.http.HttpFutureCommand;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted; import com.google.inject.assistedinject.Assisted;
/** /**
@ -40,15 +40,14 @@ import com.google.inject.assistedinject.Assisted;
* @author Adrian Cole * @author Adrian Cole
*/ */
public class HttpNioFutureCommandConnectionHandle extends public class HttpNioFutureCommandConnectionHandle extends
FutureCommandConnectionHandle<NHttpConnection, HttpFutureCommand<?>> { FutureCommandConnectionHandle<URI, NHttpConnection, HttpFutureCommand<?>> {
@Inject
public HttpNioFutureCommandConnectionHandle(
BlockingQueue<NHttpConnection> available, Semaphore maxConnections,
@Assisted NHttpConnection conn,
@Assisted HttpFutureCommand<?> command) throws InterruptedException {
super(maxConnections, command, conn, available);
// currently not injected as we want to ensure we share the correct objects with the pool
public HttpNioFutureCommandConnectionHandle(Semaphore maxConnections,
BlockingQueue<NHttpConnection> available, @Assisted URI endPoint,
@Assisted HttpFutureCommand<?> command, @Assisted NHttpConnection conn)
throws InterruptedException {
super(maxConnections, available, endPoint, command, conn);
} }
public void startConnection() { public void startConnection() {

164
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/pool/HttpNioFutureCommandConnectionPool.java
View File

@ -23,9 +23,20 @@
*/ */
package org.jclouds.http.httpnio.pool; package org.jclouds.http.httpnio.pool;
import com.google.inject.Inject; import java.io.IOException;
import com.google.inject.name.Named; import java.net.InetSocketAddress;
import java.net.URI;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.net.ssl.SSLContext;
import org.apache.http.HttpException; import org.apache.http.HttpException;
import org.apache.http.impl.nio.DefaultClientIOEventDispatch;
import org.apache.http.impl.nio.SSLClientIOEventDispatch;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor; import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.NHttpConnection; import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler; import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
@ -34,15 +45,16 @@ import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.IOReactorStatus; import org.apache.http.nio.reactor.IOReactorStatus;
import org.apache.http.nio.reactor.SessionRequest; import org.apache.http.nio.reactor.SessionRequest;
import org.apache.http.nio.reactor.SessionRequestCallback; import org.apache.http.nio.reactor.SessionRequestCallback;
import org.apache.http.params.HttpParams;
import org.jclouds.command.FutureCommand; import org.jclouds.command.FutureCommand;
import org.jclouds.command.pool.FutureCommandConnectionHandle; import org.jclouds.command.pool.FutureCommandConnectionHandle;
import org.jclouds.command.pool.FutureCommandConnectionPool; import org.jclouds.command.pool.FutureCommandConnectionPool;
import org.jclouds.command.pool.PoolConstants; import org.jclouds.command.pool.PoolConstants;
import org.jclouds.http.HttpFutureCommand; import org.jclouds.http.HttpFutureCommand;
import java.io.IOException; import com.google.inject.Inject;
import java.net.InetSocketAddress; import com.google.inject.assistedinject.Assisted;
import java.util.concurrent.*; import com.google.inject.name.Named;
/** /**
* Connection Pool for HTTP requests that utilizes Apache HTTPNio * Connection Pool for HTTP requests that utilizes Apache HTTPNio
@ -50,8 +62,8 @@ import java.util.concurrent.*;
* @author Adrian Cole * @author Adrian Cole
*/ */
public class HttpNioFutureCommandConnectionPool extends public class HttpNioFutureCommandConnectionPool extends
FutureCommandConnectionPool<NHttpConnection, HttpFutureCommand<?>> FutureCommandConnectionPool<URI, NHttpConnection, HttpFutureCommand<?>> implements
implements EventListener { EventListener {
private final NHttpClientConnectionPoolSessionRequestCallback sessionCallback; private final NHttpClientConnectionPoolSessionRequestCallback sessionCallback;
private final DefaultConnectingIOReactor ioReactor; private final DefaultConnectingIOReactor ioReactor;
@ -59,29 +71,41 @@ public class HttpNioFutureCommandConnectionPool extends
private final InetSocketAddress target; private final InetSocketAddress target;
private final int maxSessionFailures; private final int maxSessionFailures;
public static interface Factory extends
FutureCommandConnectionPool.Factory<URI, NHttpConnection, HttpFutureCommand<?>> {
HttpNioFutureCommandConnectionPool create(URI endPoint);
}
@Inject @Inject
public HttpNioFutureCommandConnectionPool( public HttpNioFutureCommandConnectionPool(ExecutorService executor, Semaphore allConnections,
ExecutorService executor,
Semaphore allConnections,
BlockingQueue<HttpFutureCommand<?>> commandQueue, BlockingQueue<HttpFutureCommand<?>> commandQueue,
BlockingQueue<NHttpConnection> available, BlockingQueue<NHttpConnection> available, AsyncNHttpClientHandler clientHandler,
AsyncNHttpClientHandler clientHandler, DefaultConnectingIOReactor ioReactor, HttpParams params,
DefaultConnectingIOReactor ioReactor,
IOEventDispatch dispatch,
FutureCommandConnectionHandleFactory requestHandleFactory,
InetSocketAddress target,
@Named(PoolConstants.PROPERTY_POOL_MAX_CONNECTION_REUSE) int maxConnectionReuse, @Named(PoolConstants.PROPERTY_POOL_MAX_CONNECTION_REUSE) int maxConnectionReuse,
@Named(PoolConstants.PROPERTY_POOL_MAX_SESSION_FAILURES) int maxSessionFailures) { @Named(PoolConstants.PROPERTY_POOL_MAX_SESSION_FAILURES) int maxSessionFailures,
super(executor, allConnections, commandQueue, requestHandleFactory, @Assisted URI endPoint) throws Exception {
maxConnectionReuse, available); super(executor, allConnections, commandQueue, maxConnectionReuse, available, endPoint);
this.ioReactor = ioReactor; this.ioReactor = ioReactor;
this.dispatch = dispatch; this.dispatch = endPoint.getScheme().equals("https") ? provideSSLClientEventDispatch(
this.target = target; clientHandler, params) : provideClientEventDispatch(clientHandler, params);
this.maxSessionFailures = maxSessionFailures; this.maxSessionFailures = maxSessionFailures;
this.sessionCallback = new NHttpClientConnectionPoolSessionRequestCallback(); this.sessionCallback = new NHttpClientConnectionPoolSessionRequestCallback();
this.target = new InetSocketAddress(getEndPoint().getHost(), getEndPoint().getPort());
clientHandler.setEventListener(this); clientHandler.setEventListener(this);
} }
public static IOEventDispatch provideSSLClientEventDispatch(AsyncNHttpClientHandler handler,
HttpParams params) throws Exception {
SSLContext context = SSLContext.getInstance("TLS");
context.init(null, null, null);
return new SSLClientIOEventDispatch(handler, context, params);
}
public static IOEventDispatch provideClientEventDispatch(AsyncNHttpClientHandler handler,
HttpParams params) throws Exception {
return new DefaultClientIOEventDispatch(handler, params);
}
@Override @Override
public void start() { public void start() {
synchronized (this.statusLock) { synchronized (this.statusLock) {
@ -119,10 +143,8 @@ public class HttpNioFutureCommandConnectionPool extends
@Override @Override
public void shutdownConnection(NHttpConnection conn) { public void shutdownConnection(NHttpConnection conn) {
if (conn.getMetrics().getRequestCount() >= maxConnectionReuse) if (conn.getMetrics().getRequestCount() >= maxConnectionReuse)
logger.debug( logger.debug("%1$s - %2$d - closing connection due to overuse %1$s/%2$s", conn, conn
"%1$s - %2$d - closing connection due to overuse %1$s/%2$s", .hashCode(), conn.getMetrics().getRequestCount(), maxConnectionReuse);
conn, conn.hashCode(), conn.getMetrics().getRequestCount(),
maxConnectionReuse);
if (conn.getStatus() == NHttpConnection.ACTIVE) { if (conn.getStatus() == NHttpConnection.ACTIVE) {
try { try {
conn.shutdown(); conn.shutdown();
@ -160,84 +182,70 @@ public class HttpNioFutureCommandConnectionPool extends
@Override @Override
protected void associateHandleWithConnection( protected void associateHandleWithConnection(
FutureCommandConnectionHandle<NHttpConnection, HttpFutureCommand<?>> handle, FutureCommandConnectionHandle<URI, NHttpConnection, HttpFutureCommand<?>> handle,
NHttpConnection connection) { NHttpConnection connection) {
connection.getContext().setAttribute("command-handle", handle); connection.getContext().setAttribute("command-handle", handle);
} }
@Override @Override
protected HttpNioFutureCommandConnectionHandle getHandleFromConnection( protected HttpNioFutureCommandConnectionHandle getHandleFromConnection(NHttpConnection connection) {
NHttpConnection connection) { return (HttpNioFutureCommandConnectionHandle) connection.getContext().getAttribute(
return (HttpNioFutureCommandConnectionHandle) connection.getContext() "command-handle");
.getAttribute("command-handle");
} }
class NHttpClientConnectionPoolSessionRequestCallback implements class NHttpClientConnectionPoolSessionRequestCallback implements SessionRequestCallback {
SessionRequestCallback {
public void completed(SessionRequest request) { public void completed(SessionRequest request) {
logger.trace("%1$s->%2$s[%3$s] - SessionRequest complete", request logger.trace("%1$s->%2$s[%3$s] - SessionRequest complete", request.getLocalAddress(),
.getLocalAddress(), request.getRemoteAddress(), request request.getRemoteAddress(), request.getAttachment());
.getAttachment());
} }
public void cancelled(SessionRequest request) { public void cancelled(SessionRequest request) {
logger.trace("%1$s->%2$s[%3$s] - SessionRequest cancelled", request logger.trace("%1$s->%2$s[%3$s] - SessionRequest cancelled", request.getLocalAddress(),
.getLocalAddress(), request.getRemoteAddress(), request request.getRemoteAddress(), request.getAttachment());
.getAttachment());
releaseConnectionAndCancelResponse(request); releaseConnectionAndCancelResponse(request);
} }
@SuppressWarnings("unchecked")
private void releaseConnectionAndCancelResponse(SessionRequest request) { private void releaseConnectionAndCancelResponse(SessionRequest request) {
allConnections.release(); allConnections.release();
FutureCommand<?, ?, ?> frequest = (FutureCommand<?, ?, ?>) request FutureCommand<URI, ?, ?, ?> frequest = (FutureCommand<URI, ?, ?, ?>) request
.getAttachment(); .getAttachment();
if (frequest != null) { if (frequest != null) {
logger.error("%1$s->%2$s[%3$s] - Cancelling FutureCommand", logger.error("%1$s->%2$s[%3$s] - Cancelling FutureCommand", request.getLocalAddress(),
request.getLocalAddress(), request.getRemoteAddress(), request.getRemoteAddress(), frequest);
frequest);
frequest.cancel(true); frequest.cancel(true);
} }
} }
private void releaseConnectionAndSetResponseException( private void releaseConnectionAndSetResponseException(SessionRequest request, Exception e) {
SessionRequest request, Exception e) {
allConnections.release(); allConnections.release();
HttpFutureCommand<?> frequest = (HttpFutureCommand<?>) request HttpFutureCommand<?> frequest = (HttpFutureCommand<?>) request.getAttachment();
.getAttachment();
if (frequest != null) { if (frequest != null) {
logger.error(e, logger.error(e, "%1$s->%2$s[%3$s] - Setting Exception on FutureCommand", request
"%1$s->%2$s[%3$s] - Setting Exception on FutureCommand", .getLocalAddress(), request.getRemoteAddress(), frequest);
request.getLocalAddress(), request.getRemoteAddress(),
frequest);
frequest.setException(e); frequest.setException(e);
} }
} }
public void failed(SessionRequest request) { public void failed(SessionRequest request) {
int count = currentSessionFailures.getAndIncrement(); int count = currentSessionFailures.getAndIncrement();
logger.warn("%1$s->%2$s[%3$s] - SessionRequest failed", request logger.warn("%1$s->%2$s[%3$s] - SessionRequest failed", request.getLocalAddress(), request
.getLocalAddress(), request.getRemoteAddress(), request .getRemoteAddress(), request.getAttachment());
.getAttachment()); releaseConnectionAndSetResponseException(request, request.getException());
releaseConnectionAndSetResponseException(request, request
.getException());
if (count >= maxSessionFailures) { if (count >= maxSessionFailures) {
logger logger.error(request.getException(),
.error(
request.getException(),
"%1$s->%2$s[%3$s] - SessionRequest failures: %4$s, Disabling pool for %5$s", "%1$s->%2$s[%3$s] - SessionRequest failures: %4$s, Disabling pool for %5$s",
request.getLocalAddress(), request request.getLocalAddress(), request.getRemoteAddress(), maxSessionFailures,
.getRemoteAddress(), target);
maxSessionFailures, target);
exception.set(request.getException()); exception.set(request.getException());
} }
} }
public void timeout(SessionRequest request) { public void timeout(SessionRequest request) {
logger.warn("%1$s->%2$s[%3$s] - SessionRequest timeout", request logger.warn("%1$s->%2$s[%3$s] - SessionRequest timeout", request.getLocalAddress(),
.getLocalAddress(), request.getRemoteAddress(), request request.getRemoteAddress(), request.getAttachment());
.getAttachment());
releaseConnectionAndCancelResponse(request); releaseConnectionAndCancelResponse(request);
} }
@ -250,8 +258,8 @@ public class HttpNioFutureCommandConnectionPool extends
} }
public void connectionTimeout(NHttpConnection conn) { public void connectionTimeout(NHttpConnection conn) {
String message = String.format("%1$s - %2$d - timeout %2$d", conn, conn String message = String.format("%1$s - %2$d - timeout %2$d", conn, conn.hashCode(), conn
.hashCode(), conn.getSocketTimeout()); .getSocketTimeout());
logger.warn(message); logger.warn(message);
resubmitIfRequestIsReplayable(conn, new TimeoutException(message)); resubmitIfRequestIsReplayable(conn, new TimeoutException(message));
} }
@ -261,27 +269,29 @@ public class HttpNioFutureCommandConnectionPool extends
} }
public void fatalIOException(IOException ex, NHttpConnection conn) { public void fatalIOException(IOException ex, NHttpConnection conn) {
logger.error(ex, "%3$s-%1$s{%2$d} - io error", conn, conn.hashCode(), logger.error(ex, "%3$s-%1$s{%2$d} - io error", conn, conn.hashCode(), target);
target);
resubmitIfRequestIsReplayable(conn, ex); resubmitIfRequestIsReplayable(conn, ex);
} }
public void fatalProtocolException(HttpException ex, NHttpConnection conn) { public void fatalProtocolException(HttpException ex, NHttpConnection conn) {
logger.error(ex, "%3$s-%1$s{%2$d} - http error", conn, conn.hashCode(), logger.error(ex, "%3$s-%1$s{%2$d} - http error", conn, conn.hashCode(), target);
target);
setExceptionOnCommand(conn, ex); setExceptionOnCommand(conn, ex);
} }
public static interface FutureCommandConnectionHandleFactory
extends
FutureCommandConnectionPool.FutureCommandConnectionHandleFactory<NHttpConnection, HttpFutureCommand<?>> {
HttpNioFutureCommandConnectionHandle create(
HttpFutureCommand<?> command, NHttpConnection conn);
}
@Override @Override
protected boolean isReplayable(HttpFutureCommand<?> command) { protected boolean isReplayable(HttpFutureCommand<?> command) {
return command.getRequest().isReplayable(); return command.getRequest().isReplayable();
} }
@Override
protected FutureCommandConnectionHandle<URI, NHttpConnection, HttpFutureCommand<?>> createHandle(
HttpFutureCommand<?> command, NHttpConnection conn) {
try {
return new HttpNioFutureCommandConnectionHandle(allConnections, available, endPoint,
command, conn);
} catch (InterruptedException e) {
throw new RuntimeException("Interrupted creating a handle to " + conn, e);
}
}
} }

51
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/pool/HttpNioFutureCommandExecutionHandler.java
View File

@ -39,8 +39,8 @@ import org.jclouds.http.HttpErrorHandler;
import org.jclouds.http.HttpFutureCommand; import org.jclouds.http.HttpFutureCommand;
import org.jclouds.http.HttpRequest; import org.jclouds.http.HttpRequest;
import org.jclouds.http.HttpRetryHandler; import org.jclouds.http.HttpRetryHandler;
import org.jclouds.http.handlers.BackoffLimitedRetryHandler; import org.jclouds.http.handlers.DelegatingErrorHandler;
import org.jclouds.http.handlers.CloseContentAndSetExceptionHandler; import org.jclouds.http.handlers.DelegatingRetryHandler;
import org.jclouds.http.httpnio.util.HttpNioUtils; import org.jclouds.http.httpnio.util.HttpNioUtils;
import org.jclouds.logging.Logger; import org.jclouds.logging.Logger;
@ -56,24 +56,28 @@ public class HttpNioFutureCommandExecutionHandler implements NHttpRequestExecuti
@Resource @Resource
protected Logger logger = Logger.NULL; protected Logger logger = Logger.NULL;
private final ConsumingNHttpEntityFactory entityFactory; private final ConsumingNHttpEntityFactory entityFactory;
private final BlockingQueue<HttpFutureCommand<?>> commandQueue;
/**
* inputOnly: nothing is taken from this queue.
*/
private final BlockingQueue<HttpFutureCommand<?>> resubmitQueue;
@Inject(optional = true) @Inject(optional = true)
private HttpErrorHandler serverErrorHandler = new CloseContentAndSetExceptionHandler(); private HttpRetryHandler retryHandler = new DelegatingRetryHandler();
@Inject(optional = true) @Inject(optional = true)
protected HttpRetryHandler httpRetryHandler = new BackoffLimitedRetryHandler(5); private HttpErrorHandler errorHandler = new DelegatingErrorHandler();
public interface ConsumingNHttpEntityFactory {
public ConsumingNHttpEntity create(HttpEntity httpEntity);
}
@Inject @Inject
public HttpNioFutureCommandExecutionHandler(ConsumingNHttpEntityFactory entityFactory, public HttpNioFutureCommandExecutionHandler(ConsumingNHttpEntityFactory entityFactory,
ExecutorService executor, BlockingQueue<HttpFutureCommand<?>> commandQueue) { ExecutorService executor, BlockingQueue<HttpFutureCommand<?>> resubmitQueue) {
this.executor = executor; this.executor = executor;
this.entityFactory = entityFactory; this.entityFactory = entityFactory;
this.commandQueue = commandQueue; this.resubmitQueue = resubmitQueue;
}
public interface ConsumingNHttpEntityFactory {
public ConsumingNHttpEntity create(HttpEntity httpEntity);
} }
public void initalizeContext(HttpContext context, Object attachment) { public void initalizeContext(HttpContext context, Object attachment) {
@ -82,8 +86,8 @@ public class HttpNioFutureCommandExecutionHandler implements NHttpRequestExecuti
public HttpEntityEnclosingRequest submitRequest(HttpContext context) { public HttpEntityEnclosingRequest submitRequest(HttpContext context) {
HttpFutureCommand<?> command = (HttpFutureCommand<?>) context.removeAttribute("command"); HttpFutureCommand<?> command = (HttpFutureCommand<?>) context.removeAttribute("command");
if (command != null) { if (command != null) {
HttpRequest object = command.getRequest(); HttpRequest request = command.getRequest();
return HttpNioUtils.convertToApacheRequest(object); return HttpNioUtils.convertToApacheRequest(request);
} }
return null; return null;
@ -102,24 +106,13 @@ public class HttpNioFutureCommandExecutionHandler implements NHttpRequestExecuti
HttpFutureCommand<?> command = handle.getCommand(); HttpFutureCommand<?> command = handle.getCommand();
org.jclouds.http.HttpResponse response = HttpNioUtils org.jclouds.http.HttpResponse response = HttpNioUtils
.convertToJavaCloudsResponse(apacheResponse); .convertToJavaCloudsResponse(apacheResponse);
int statusCode = response.getStatusCode();
int code = response.getStatusCode(); if (statusCode >= 300) {
if (code >= 500) { if (retryHandler.shouldRetryRequest(command, response)) {
boolean retryRequest = false; resubmitQueue.add(command);
try {
retryRequest = httpRetryHandler.shouldRetryRequest(command, response);
} catch (InterruptedException ie) {
// TODO: Add interrupt exception to command and abort?
}
if (retryRequest) {
commandQueue.add(command);
} else { } else {
serverErrorHandler.handle(command, response); errorHandler.handleError(command, response);
} }
} else if (code >= 400 && code < 500) {
serverErrorHandler.handle(command, response);
} else if (code >= 300 && code < 400) {
serverErrorHandler.handle(command, response);
} else { } else {
processResponse(response, command); processResponse(response, command);
} }

25
extensions/httpnio/src/main/java/org/jclouds/http/httpnio/util/HttpNioUtils.java
View File

@ -41,25 +41,28 @@ import org.jclouds.http.HttpRequest;
import org.jclouds.http.HttpResponse; import org.jclouds.http.HttpResponse;
public class HttpNioUtils { public class HttpNioUtils {
public static HttpEntityEnclosingRequest convertToApacheRequest(HttpRequest object) { public static HttpEntityEnclosingRequest convertToApacheRequest(HttpRequest request) {
BasicHttpEntityEnclosingRequest apacheRequest = new BasicHttpEntityEnclosingRequest(object BasicHttpEntityEnclosingRequest apacheRequest = new BasicHttpEntityEnclosingRequest(request
.getMethod().toString(), object.getUri(), HttpVersion.HTTP_1_1); .getMethod().toString(), request.getUri(), HttpVersion.HTTP_1_1);
Object content = object.getPayload(); Object content = request.getPayload();
// Since we may remove headers, ensure they are added to the apache // Since we may remove headers, ensure they are added to the apache
// request after this block // request after this block
if (content != null) { if (content != null) {
long contentLength = Long.parseLong(object String lengthString = request.getFirstHeaderOrNull(HttpHeaders.CONTENT_LENGTH);
.getFirstHeaderOrNull(HttpHeaders.CONTENT_LENGTH)); if (lengthString == null) {
object.getHeaders().removeAll(HttpHeaders.CONTENT_LENGTH); throw new IllegalStateException("no Content-Length header on request: " + apacheRequest);
String contentType = object.getFirstHeaderOrNull(HttpHeaders.CONTENT_TYPE); }
object.getHeaders().removeAll(HttpHeaders.CONTENT_TYPE); long contentLength = Long.parseLong(lengthString);
String contentType = request.getFirstHeaderOrNull(HttpHeaders.CONTENT_TYPE);
addEntityForContent(apacheRequest, content, contentType, contentLength); addEntityForContent(apacheRequest, content, contentType, contentLength);
} }
for (String header : object.getHeaders().keySet()) { for (String header : request.getHeaders().keySet()) {
for (String value : object.getHeaders().get(header)) for (String value : request.getHeaders().get(header))
// apache automatically tries to add content length header
if (!header.equals(HttpHeaders.CONTENT_LENGTH))
apacheRequest.addHeader(header, value); apacheRequest.addHeader(header, value);
} }
return apacheRequest; return apacheRequest;

15
extensions/httpnio/src/test/java/org/jclouds/http/httpnio/pool/HttpNioConnectionPoolFutureCommandClientTest.java
View File

@ -23,10 +23,7 @@
*/ */
package org.jclouds.http.httpnio.pool; package org.jclouds.http.httpnio.pool;
import java.net.MalformedURLException;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.jclouds.command.pool.PoolConstants; import org.jclouds.command.pool.PoolConstants;
import org.jclouds.http.BaseHttpFutureCommandClientTest; import org.jclouds.http.BaseHttpFutureCommandClientTest;
@ -55,16 +52,4 @@ public class HttpNioConnectionPoolFutureCommandClientTest extends BaseHttpFuture
return new HttpNioConnectionPoolClientModule(); return new HttpNioConnectionPoolClientModule();
} }
@Override
@Test(enabled = false)
public void testGetStringRedirect() throws MalformedURLException, ExecutionException,
InterruptedException, TimeoutException {
}
@Override
@Test(enabled = false)
public void testPutRedirect() throws MalformedURLException, ExecutionException,
InterruptedException, TimeoutException {
}
} }