You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2019/09/12 08:07:44 UTC
[httpcomponents-core] 01/03: IOSession to implement ByteChannel
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
commit 01a0375ce2f41839bce122ebd2285f31098b047d
Author: Oleg Kalnichevski <ol...@apache.org>
AuthorDate: Tue Sep 10 11:40:57 2019 +0200
IOSession to implement ByteChannel
---
.../impl/nio/ClientHttpProtocolNegotiator.java | 2 +-
.../impl/nio/H2OnlyClientProtocolNegotiator.java | 2 +-
.../impl/nio/ServerHttpProtocolNegotiator.java | 2 +-
.../nio/bootstrap/H2MultiplexingRequester.java | 2 +-
.../apache/hc/core5/benchmark/HttpBenchmark.java | 54 ++++++-------
.../core5/testing/nio/ClientSessionEndpoint.java | 6 +-
.../hc/core5/testing/nio/LoggingIOSession.java | 90 +++++++++-------------
.../hc/core5/testing/nio/Http1IntegrationTest.java | 8 +-
.../http/impl/bootstrap/HttpAsyncRequester.java | 8 +-
.../hc/core5/http/nio/command/CommandSupport.java | 2 +-
.../hc/core5/reactor/AbstractIOSessionPool.java | 4 +-
.../org/apache/hc/core5/reactor/IOSession.java | 5 +-
.../org/apache/hc/core5/reactor/IOSessionImpl.java | 17 ++++
.../hc/core5/reactor/InternalDataChannel.java | 16 ++++
.../apache/hc/core5/reactor/ssl/SSLIOSession.java | 39 +++-------
.../core5/reactor/TestAbstractIOSessionPool.java | 6 +-
16 files changed, 132 insertions(+), 131 deletions(-)
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
index 7226a89..2c0f6b6 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
@@ -232,7 +232,7 @@ public class ClientHttpProtocolNegotiator implements HttpConnectionEventHandler
@Override
public boolean isOpen() {
- return !ioSession.isClosed();
+ return ioSession.isOpen();
}
@Override
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2OnlyClientProtocolNegotiator.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2OnlyClientProtocolNegotiator.java
index 5165808..3809d52 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2OnlyClientProtocolNegotiator.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2OnlyClientProtocolNegotiator.java
@@ -189,7 +189,7 @@ public class H2OnlyClientProtocolNegotiator implements HttpConnectionEventHandle
@Override
public boolean isOpen() {
- return !ioSession.isClosed();
+ return ioSession.isOpen();
}
@Override
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerHttpProtocolNegotiator.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerHttpProtocolNegotiator.java
index 2357eb1..a2776a1 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerHttpProtocolNegotiator.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerHttpProtocolNegotiator.java
@@ -242,7 +242,7 @@ public class ServerHttpProtocolNegotiator implements HttpConnectionEventHandler
@Override
public boolean isOpen() {
- return !ioSession.isClosed();
+ return ioSession.isOpen();
}
@Override
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
index 39b6d64..c77354f 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
@@ -225,7 +225,7 @@ public class H2MultiplexingRequester extends AsyncRequester{
}
}, pushHandlerFactory, cancellableDependency, context), Command.Priority.NORMAL);
- if (ioSession.isClosed()) {
+ if (!ioSession.isOpen()) {
exchangeHandler.failed(new ConnectionClosedException());
}
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java
index b12d850..e055718 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java
@@ -211,36 +211,7 @@ public class HttpBenchmark {
@Override
public ByteChannel channel() {
- return new ByteChannel() {
-
- @Override
- public int read(final ByteBuffer dst) throws IOException {
- final int bytesRead = ioSession.channel().read(dst);
- if (bytesRead > 0) {
- stats.incTotalBytesRecv(bytesRead);
- }
- return bytesRead;
- }
-
- @Override
- public int write(final ByteBuffer src) throws IOException {
- final int bytesWritten = ioSession.channel().write(src);
- if (bytesWritten > 0) {
- stats.incTotalBytesSent(bytesWritten);
- }
- return bytesWritten;
- }
-
- @Override
- public boolean isOpen() {
- return ioSession.channel().isOpen();
- }
-
- @Override
- public void close() throws IOException {
- ioSession.channel().close();
- }
- };
+ return this;
}
@Override
@@ -284,6 +255,29 @@ public class HttpBenchmark {
}
@Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int bytesRead = ioSession.read(dst);
+ if (bytesRead > 0) {
+ stats.incTotalBytesRecv(bytesRead);
+ }
+ return bytesRead;
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ final int bytesWritten = ioSession.write(src);
+ if (bytesWritten > 0) {
+ stats.incTotalBytesSent(bytesWritten);
+ }
+ return bytesWritten;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return ioSession.isOpen();
+ }
+
+ @Override
public boolean isClosed() {
return ioSession.isClosed();
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
index 3672ae1..6686f59 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
@@ -70,7 +70,7 @@ public final class ClientSessionEndpoint implements ModalCloseable {
public void execute(final Command command, final Command.Priority priority) {
ioSession.enqueue(command, priority);
- if (ioSession.isClosed()) {
+ if (!ioSession.isOpen()) {
command.cancel();
}
}
@@ -82,7 +82,7 @@ public final class ClientSessionEndpoint implements ModalCloseable {
Asserts.check(!closed.get(), "Connection is already closed");
final Command executionCommand = new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, null, context);
ioSession.enqueue(executionCommand, Command.Priority.NORMAL);
- if (ioSession.isClosed()) {
+ if (!ioSession.isOpen()) {
exchangeHandler.failed(new ConnectionClosedException());
}
}
@@ -140,7 +140,7 @@ public final class ClientSessionEndpoint implements ModalCloseable {
}
public boolean isOpen() {
- return !closed.get() && !ioSession.isClosed();
+ return !closed.get() && ioSession.isOpen();
}
@Override
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingIOSession.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingIOSession.java
index 41697f3..f1b4fd3 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingIOSession.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingIOSession.java
@@ -54,14 +54,12 @@ public class LoggingIOSession implements ProtocolIOSession {
private final Logger log;
private final Wire wireLog;
private final ProtocolIOSession session;
- private final ByteChannel channel;
public LoggingIOSession(final ProtocolIOSession session, final Logger log, final Logger wireLog) {
super();
this.session = session;
this.log = log;
this.wireLog = wireLog != null ? new Wire(wireLog, session.getId()) : null;
- this.channel = wireLog != null ? new LoggingByteChannel() : session.channel();
}
public LoggingIOSession(final ProtocolIOSession session, final Logger log) {
@@ -98,7 +96,7 @@ public class LoggingIOSession implements ProtocolIOSession {
@Override
public ByteChannel channel() {
- return this.channel;
+ return this;
}
@Override
@@ -178,6 +176,11 @@ public class LoggingIOSession implements ProtocolIOSession {
}
@Override
+ public boolean isOpen() {
+ return session.isOpen();
+ }
+
+ @Override
public void close(final CloseMode closeMode) {
if (this.log.isDebugEnabled()) {
this.log.debug(this.session + " Shutdown " + closeMode);
@@ -199,6 +202,38 @@ public class LoggingIOSession implements ProtocolIOSession {
}
@Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int bytesRead = session.read(dst);
+ if (log.isDebugEnabled()) {
+ log.debug(session + " " + bytesRead + " bytes read");
+ }
+ if (bytesRead > 0 && wireLog.isEnabled()) {
+ final ByteBuffer b = dst.duplicate();
+ final int p = b.position();
+ b.limit(p);
+ b.position(p - bytesRead);
+ wireLog.input(b);
+ }
+ return bytesRead;
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ final int byteWritten = session.write(src);
+ if (log.isDebugEnabled()) {
+ log.debug(session + " " + byteWritten + " bytes written");
+ }
+ if (byteWritten > 0 && wireLog.isEnabled()) {
+ final ByteBuffer b = src.duplicate();
+ final int p = b.position();
+ b.limit(p);
+ b.position(p - byteWritten);
+ wireLog.output(b);
+ }
+ return byteWritten;
+ }
+
+ @Override
public void updateReadTime() {
this.session.updateReadTime();
}
@@ -264,53 +299,4 @@ public class LoggingIOSession implements ProtocolIOSession {
return this.session.toString();
}
- class LoggingByteChannel implements ByteChannel {
-
- @Override
- public int read(final ByteBuffer dst) throws IOException {
- final int bytesRead = session.channel().read(dst);
- if (log.isDebugEnabled()) {
- log.debug(session + " " + bytesRead + " bytes read");
- }
- if (bytesRead > 0 && wireLog.isEnabled()) {
- final ByteBuffer b = dst.duplicate();
- final int p = b.position();
- b.limit(p);
- b.position(p - bytesRead);
- wireLog.input(b);
- }
- return bytesRead;
- }
-
- @Override
- public int write(final ByteBuffer src) throws IOException {
- final int byteWritten = session.channel().write(src);
- if (log.isDebugEnabled()) {
- log.debug(session + " " + byteWritten + " bytes written");
- }
- if (byteWritten > 0 && wireLog.isEnabled()) {
- final ByteBuffer b = src.duplicate();
- final int p = b.position();
- b.limit(p);
- b.position(p - byteWritten);
- wireLog.output(b);
- }
- return byteWritten;
- }
-
- @Override
- public void close() throws IOException {
- if (log.isDebugEnabled()) {
- log.debug(session + " Channel close");
- }
- session.channel().close();
- }
-
- @Override
- public boolean isOpen() {
- return session.channel().isOpen();
- }
-
- }
-
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java
index a19a48e..e167c38 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java
@@ -746,7 +746,7 @@ public class Http1IntegrationTest extends InternalHttp1ServerTestBase {
Assert.assertEquals(200, response1.getCode());
Assert.assertNotNull("All is well", result1.getBody());
- Assert.assertFalse(ioSession.isClosed());
+ Assert.assertTrue(ioSession.isOpen());
final HttpRequest request2 = new BasicHttpRequest(Methods.POST, createRequestURI(serverEndpoint, "/echo"));
final Future<Message<HttpResponse, String>> future2 = streamEndpoint.execute(
@@ -759,7 +759,7 @@ public class Http1IntegrationTest extends InternalHttp1ServerTestBase {
Assert.assertEquals(HttpStatus.SC_UNAUTHORIZED, response2.getCode());
Assert.assertNotNull("You shall not pass", result2.getBody());
- Assert.assertFalse(ioSession.isClosed());
+ Assert.assertTrue(ioSession.isOpen());
final HttpRequest request3 = new BasicHttpRequest(Methods.POST, createRequestURI(serverEndpoint, "/echo"));
request3.addHeader("password", "secret");
@@ -773,7 +773,7 @@ public class Http1IntegrationTest extends InternalHttp1ServerTestBase {
Assert.assertEquals(200, response3.getCode());
Assert.assertNotNull("All is well", result3.getBody());
- Assert.assertFalse(ioSession.isClosed());
+ Assert.assertTrue(ioSession.isOpen());
final HttpRequest request4 = new BasicHttpRequest(Methods.POST, createRequestURI(serverEndpoint, "/echo"));
final Future<Message<HttpResponse, String>> future4 = streamEndpoint.execute(
@@ -786,7 +786,7 @@ public class Http1IntegrationTest extends InternalHttp1ServerTestBase {
Assert.assertEquals(HttpStatus.SC_UNAUTHORIZED, response4.getCode());
Assert.assertNotNull("You shall not pass", result4.getBody());
- Assert.assertTrue(ioSession.isClosed());
+ Assert.assertFalse(ioSession.isOpen());
}
@Test
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
index c10a721..7b7c7a9 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
@@ -182,7 +182,7 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
public void completed(final PoolEntry<HttpHost, IOSession> poolEntry) {
final AsyncClientEndpoint endpoint = new InternalAsyncClientEndpoint(poolEntry);
final IOSession ioSession = poolEntry.getConnection();
- if (ioSession != null && ioSession.isClosed()) {
+ if (ioSession != null && !ioSession.isOpen()) {
poolEntry.discardConnection(CloseMode.IMMEDIATE);
}
if (poolEntry.hasConnection()) {
@@ -433,7 +433,7 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
throw new IllegalStateException("I/O session is invalid");
}
ioSession.enqueue(new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, null, context), Command.Priority.NORMAL);
- if (ioSession.isClosed()) {
+ if (!ioSession.isOpen()) {
exchangeHandler.failed(new ConnectionClosedException());
}
}
@@ -443,7 +443,7 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
final PoolEntry<HttpHost, IOSession> poolEntry = poolEntryRef.get();
if (poolEntry != null) {
final IOSession ioSession = poolEntry.getConnection();
- if (ioSession != null && !ioSession.isClosed()) {
+ if (ioSession != null && ioSession.isOpen()) {
return true;
}
}
@@ -455,7 +455,7 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
final PoolEntry<HttpHost, IOSession> poolEntry = poolEntryRef.getAndSet(null);
if (poolEntry != null) {
final IOSession ioSession = poolEntry.getConnection();
- connPool.release(poolEntry, ioSession != null && !ioSession.isClosed());
+ connPool.release(poolEntry, ioSession != null && ioSession.isOpen());
}
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/command/CommandSupport.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/command/CommandSupport.java
index c4eb16c..c2facea 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/command/CommandSupport.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/command/CommandSupport.java
@@ -72,7 +72,7 @@ public final class CommandSupport {
if (command instanceof RequestExecutionCommand) {
final AsyncClientExchangeHandler exchangeHandler = ((RequestExecutionCommand) command).getExchangeHandler();
try {
- if (ioSession.isClosed()) {
+ if (!ioSession.isOpen()) {
exchangeHandler.failed(new ConnectionClosedException());
} else {
exchangeHandler.cancel();
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
index fa743b7..33528ac 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
@@ -189,7 +189,7 @@ public abstract class AbstractIOSessionPool<T> implements ModalCloseable {
closeSession(poolEntry.session, CloseMode.GRACEFUL);
poolEntry.session = null;
}
- if (poolEntry.session != null && poolEntry.session.isClosed()) {
+ if (poolEntry.session != null && !poolEntry.session.isOpen()) {
poolEntry.session = null;
}
if (poolEntry.session != null) {
@@ -251,7 +251,7 @@ public abstract class AbstractIOSessionPool<T> implements ModalCloseable {
synchronized (poolEntry) {
if (poolEntry.session != null) {
callback.execute(poolEntry.session);
- if (poolEntry.session.isClosed()) {
+ if (!poolEntry.session.isOpen()) {
poolEntry.session = null;
}
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java
index 30317da..30cd22d 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java
@@ -51,7 +51,7 @@ import org.apache.hc.core5.util.Timeout;
*
* @since 4.0
*/
-public interface IOSession extends ModalCloseable, Identifiable {
+public interface IOSession extends ByteChannel, ModalCloseable, Identifiable {
int ACTIVE = 0;
int CLOSING = 1;
@@ -164,7 +164,10 @@ public interface IOSession extends ModalCloseable, Identifiable {
*
* @return {@code true} if the session has been terminated,
* {@code false} otherwise.
+ *
+ * @deprecated Use {@link #isOpen()}
*/
+ @Deprecated
boolean isClosed();
/**
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java
index 579d3bb..a4ed30c 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java
@@ -27,8 +27,10 @@
package org.apache.hc.core5.reactor;
+import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketException;
+import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
@@ -180,6 +182,16 @@ class IOSessionImpl implements IOSession {
}
@Override
+ public int read(final ByteBuffer dst) throws IOException {
+ return this.channel.read(dst);
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ return this.channel.write(src);
+ }
+
+ @Override
public void updateReadTime() {
lastReadTime = System.currentTimeMillis();
lastEventTime = lastReadTime;
@@ -221,6 +233,11 @@ class IOSessionImpl implements IOSession {
}
@Override
+ public boolean isOpen() {
+ return this.status.get() == ACTIVE && this.channel.isOpen();
+ }
+
+ @Override
public void close() {
close(CloseMode.GRACEFUL);
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalDataChannel.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalDataChannel.java
index 53b5231..f2c0fe3 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalDataChannel.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalDataChannel.java
@@ -29,6 +29,7 @@ package org.apache.hc.core5.reactor;
import java.io.IOException;
import java.net.SocketAddress;
+import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectionKey;
import java.util.Queue;
@@ -310,6 +311,11 @@ final class InternalDataChannel extends InternalChannel implements ProtocolIOSes
}
@Override
+ public boolean isOpen() {
+ return getSessionImpl().isOpen();
+ }
+
+ @Override
public boolean isClosed() {
return getSessionImpl().isClosed();
}
@@ -375,6 +381,16 @@ final class InternalDataChannel extends InternalChannel implements ProtocolIOSes
}
@Override
+ public int read(final ByteBuffer dst) throws IOException {
+ return getSessionImpl().read(dst);
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ return getSessionImpl().write(src);
+ }
+
+ @Override
public void updateReadTime() {
ioSession.updateReadTime();
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/ssl/SSLIOSession.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/ssl/SSLIOSession.java
index 708978b..e75da59 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/ssl/SSLIOSession.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/ssl/SSLIOSession.java
@@ -78,7 +78,6 @@ public class SSLIOSession implements IOSession {
private final SSLManagedBuffer inEncrypted;
private final SSLManagedBuffer outEncrypted;
private final SSLManagedBuffer inPlain;
- private final ByteChannel channel;
private final SSLSessionInitializer initializer;
private final SSLSessionVerifier verifier;
private final Callback<SSLIOSession> connectedCallback;
@@ -147,29 +146,6 @@ public class SSLIOSession implements IOSession {
// Allocate buffers for application (unencrypted) data
final int appBufferSize = sslSession.getApplicationBufferSize();
this.inPlain = SSLManagedBuffer.create(sslBufferMode, appBufferSize);
- this.channel = new ByteChannel() {
-
- @Override
- public int write(final ByteBuffer src) throws IOException {
- return SSLIOSession.this.writePlain(src);
- }
-
- @Override
- public int read(final ByteBuffer dst) throws IOException {
- return SSLIOSession.this.readPlain(dst);
- }
-
- @Override
- public void close() throws IOException {
- SSLIOSession.this.close();
- }
-
- @Override
- public boolean isOpen() {
- return !SSLIOSession.this.isClosed();
- }
-
- };
this.bytesReadCount = new AtomicLong(0);
this.connectTimeout = connectTimeout;
}
@@ -576,7 +552,7 @@ public class SSLIOSession implements IOSession {
public void outboundTransport() throws IOException {
this.session.getLock().lock();
try {
- if (this.session.isClosed()) {
+ if (!this.session.isOpen()) {
return;
}
sendEncryptedData();
@@ -601,7 +577,8 @@ public class SSLIOSession implements IOSession {
return this.sslEngine.isOutboundDone();
}
- private int writePlain(final ByteBuffer src) throws IOException {
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
Args.notNull(src, "Byte buffer");
this.session.getLock().lock();
try {
@@ -622,7 +599,8 @@ public class SSLIOSession implements IOSession {
}
}
- private int readPlain(final ByteBuffer dst) {
+ @Override
+ public int read(final ByteBuffer dst) {
Args.notNull(dst, "Byte buffer");
this.session.getLock().lock();
try {
@@ -666,6 +644,11 @@ public class SSLIOSession implements IOSession {
}
@Override
+ public boolean isOpen() {
+ return this.status == ACTIVE && this.session.isOpen();
+ }
+
+ @Override
public void close() {
close(CloseMode.GRACEFUL);
}
@@ -742,7 +725,7 @@ public class SSLIOSession implements IOSession {
@Override
public ByteChannel channel() {
- return this.channel;
+ return this;
}
@Override
diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestAbstractIOSessionPool.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestAbstractIOSessionPool.java
index 267075d..fe87e61 100644
--- a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestAbstractIOSessionPool.java
+++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestAbstractIOSessionPool.java
@@ -89,6 +89,8 @@ public class TestAbstractIOSessionPool {
}).when(impl).validateSession(ArgumentMatchers.<IOSession>any(), ArgumentMatchers.<Callback<Boolean>>any());
+ Mockito.when(ioSession1.isOpen()).thenReturn(true);
+
final Future<IOSession> future1 = impl.getSession("somehost", Timeout.ofSeconds(123L), null);
Assert.assertThat(future1, CoreMatchers.notNullValue());
Assert.assertThat(future1.isDone(), CoreMatchers.equalTo(false));
@@ -250,7 +252,7 @@ public class TestAbstractIOSessionPool {
Assert.assertThat(entry1, CoreMatchers.notNullValue());
entry1.session = ioSession1;
- Mockito.when(ioSession1.isClosed()).thenReturn(false);
+ Mockito.when(ioSession1.isOpen()).thenReturn(true);
Mockito.doAnswer(new Answer() {
@Override
@@ -276,7 +278,7 @@ public class TestAbstractIOSessionPool {
Assert.assertThat(entry1, CoreMatchers.notNullValue());
entry1.session = ioSession1;
- Mockito.when(ioSession1.isClosed()).thenReturn(true);
+ Mockito.when(ioSession1.isOpen()).thenReturn(false);
impl.getSession("somehost", Timeout.ofSeconds(123L), null);