CARVIEW |
Select Language
HTTP/2 200
date: Wed, 23 Jul 2025 16:28:19 GMT
content-type: text/html; charset=utf-8
cache-control: max-age=0, private, must-revalidate
content-security-policy: default-src 'none'; base-uri 'self'; child-src github.githubassets.com github.com/assets-cdn/worker/ github.com/assets/ gist.github.com/assets-cdn/worker/; connect-src 'self' uploads.github.com www.githubstatus.com collector.github.com raw.githubusercontent.com api.github.com github-cloud.s3.amazonaws.com github-production-repository-file-5c1aeb.s3.amazonaws.com github-production-upload-manifest-file-7fdce7.s3.amazonaws.com github-production-user-asset-6210df.s3.amazonaws.com *.rel.tunnels.api.visualstudio.com wss://*.rel.tunnels.api.visualstudio.com objects-origin.githubusercontent.com copilot-proxy.githubusercontent.com proxy.individual.githubcopilot.com proxy.business.githubcopilot.com proxy.enterprise.githubcopilot.com *.actions.githubusercontent.com wss://*.actions.githubusercontent.com productionresultssa0.blob.core.windows.net/ productionresultssa1.blob.core.windows.net/ productionresultssa2.blob.core.windows.net/ productionresultssa3.blob.core.windows.net/ productionresultssa4.blob.core.windows.net/ productionresultssa5.blob.core.windows.net/ productionresultssa6.blob.core.windows.net/ productionresultssa7.blob.core.windows.net/ productionresultssa8.blob.core.windows.net/ productionresultssa9.blob.core.windows.net/ productionresultssa10.blob.core.windows.net/ productionresultssa11.blob.core.windows.net/ productionresultssa12.blob.core.windows.net/ productionresultssa13.blob.core.windows.net/ productionresultssa14.blob.core.windows.net/ productionresultssa15.blob.core.windows.net/ productionresultssa16.blob.core.windows.net/ productionresultssa17.blob.core.windows.net/ productionresultssa18.blob.core.windows.net/ productionresultssa19.blob.core.windows.net/ github-production-repository-image-32fea6.s3.amazonaws.com github-production-release-asset-2e65be.s3.amazonaws.com insights.github.com wss://alive.github.com api.githubcopilot.com api.individual.githubcopilot.com api.business.githubcopilot.com api.enterprise.githubcopilot.com; font-src github.githubassets.com; form-action 'self' github.com gist.github.com copilot-workspace.githubnext.com objects-origin.githubusercontent.com; frame-ancestors 'none'; frame-src viewscreen.githubusercontent.com notebooks.githubusercontent.com; img-src 'self' data: blob: github.githubassets.com media.githubusercontent.com camo.githubusercontent.com identicons.github.com avatars.githubusercontent.com private-avatars.githubusercontent.com github-cloud.s3.amazonaws.com objects.githubusercontent.com release-assets.githubusercontent.com secured-user-images.githubusercontent.com/ user-images.githubusercontent.com/ private-user-images.githubusercontent.com opengraph.githubassets.com copilotprodattachments.blob.core.windows.net/github-production-copilot-attachments/ github-production-user-asset-6210df.s3.amazonaws.com customer-stories-feed.github.com spotlights-feed.github.com objects-origin.githubusercontent.com *.githubusercontent.com; manifest-src 'self'; media-src github.com user-images.githubusercontent.com/ secured-user-images.githubusercontent.com/ private-user-images.githubusercontent.com github-production-user-asset-6210df.s3.amazonaws.com gist.github.com; script-src github.githubassets.com; style-src 'unsafe-inline' github.githubassets.com; upgrade-insecure-requests; worker-src github.githubassets.com github.com/assets-cdn/worker/ github.com/assets/ gist.github.com/assets-cdn/worker/
link: ; rel=preload; as=fetch; crossorigin=use-credentials
referrer-policy: no-referrer-when-downgrade
server-timing: issue_layout-fragment;desc="issue_layout fragment";dur=277.467833,issue_conversation_content-fragment;desc="issue_conversation_content fragment";dur=583.128714,issue_conversation_sidebar-fragment;desc="issue_conversation_sidebar fragment";dur=47.540212,nginx;desc="NGINX";dur=1.004498,glb;desc="GLB";dur=101.255461
strict-transport-security: max-age=31536000; includeSubdomains; preload
vary: X-PJAX, X-PJAX-Container, Turbo-Visit, Turbo-Frame, X-Requested-With, Accept,Accept-Encoding, Accept, X-Requested-With
x-content-type-options: nosniff
x-frame-options: deny
x-voltron-version: fd8fbbc
x-xss-protection: 0
server: github.com
content-encoding: gzip
accept-ranges: bytes
set-cookie: _gh_sess=gnFrb2z2BOy0t0paIZ6%2FeOuQezID73AJ46BkzXd%2BD8ry6hPTGX2%2BTl%2BsbUBkEHM0aWVDC0kMZ%2BOq2%2BXju5ug8gsP0TfV4XTNCnVWj2HbJuBKmWzEYWhPyD0f37yoSjJf7l0eTnVkwxgORAY7lemQWoyr%2BilY8HDsNGCqUyGkwUV1W1cMlQJIiv3AOY%2BYZ0Jazo8ez0H%2BzP8QzLr2EhlNbGzd%2FQuD2g954cHIIPfjPFKpPlT1vGRiyg0ReTktfIoQB%2FGpTfMvW473RPbI47Lj9A%3D%3D--l07nkWAgBHG4e6wz--0sJkx8RGfl%2ButOrPD1MdHg%3D%3D; Path=/; HttpOnly; Secure; SameSite=Lax
set-cookie: _octo=GH1.1.808441753.1753288098; Path=/; Domain=github.com; Expires=Thu, 23 Jul 2026 16:28:18 GMT; Secure; SameSite=Lax
set-cookie: logged_in=no; Path=/; Domain=github.com; Expires=Thu, 23 Jul 2026 16:28:18 GMT; HttpOnly; Secure; SameSite=Lax
x-github-request-id: 8D2A:36F97A:EA7281:114F849:68810DA2
TCP client API changes for 0.5.0 · Issue #317 · ReactiveX/RxNetty · GitHub
No labelsNo typeNo projectsNone yetNo branches or pull requests
Skip to content
Navigation Menu
{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 257
Closed
Milestone
Description
This issue contains the API changes to TCP client based on the issues #280 #281
Tcp Client
This client does not extend RxClient
, I am preferring delegation instead of inheritance for the clients, so all clients (HTTP/UDP/TCP) will be standalone.
All mutations to this client, creates a new instance.
/**
* A TCP client for creating TCP connections.
*
* <h2>Immutability</h2>
* An instance of this client is immutable and all mutations produce a new client instance. For this reason it is
* recommended that the mutations are done during client creation and not during connection creation to avoid repeated
* object creation overhead.
*
* @param <I> The type of objects written to this client.
* @param <O> The type of objects read from this client.
*/
public abstract class TcpClient<I, O> {
/**
* Creates a new {@link ConnectionRequest} which should be subscribed to actually connect to the target server.
*
* @return A new {@link ConnectionRequest} which either can be subscribed directly or altered in various ways
* before subscription.
*/
public abstract ConnectionRequest<I, O> createConnectionRequest();
/**
* Creates a new {@link ConnectionRequest} which should be subscribed to actually connect to the target server.
* This method overrides the default host and port configured for this client.
*
* @param host Target host to connect.
* @param port Port on the host to connect.
*
* @return A new {@link ConnectionRequest} which either can be subscribed directly or altered in various ways
* before subscription.
*/
public abstract ConnectionRequest<I, O> createConnectionRequest(String host, int port);
/**
* Creates a new client instances, inheriting all configurations from this client and adding a
* {@link ChannelOption} for the connections created by the newly created client instance.
*
* @param option Option to add.
* @param value Value for the option.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <T> TcpClient<I, O> channelOption(ChannelOption<T> option, T value);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added at the first position of the pipeline as specified by
* {@link ChannelPipeline#addFirst(String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerFirst(String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added at the first position of the pipeline as specified by
* {@link ChannelPipeline#addFirst(EventExecutorGroup, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerFirst(EventExecutorGroup group, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added at the last position of the pipeline as specified by
* {@link ChannelPipeline#addLast(String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerLast(String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added at the last position of the pipeline as specified by
* {@link ChannelPipeline#addLast(EventExecutorGroup, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerLast(EventExecutorGroup group, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added before an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addBefore(String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param baseName the name of the existing handler
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerBefore(String baseName, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added before an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addBefore(EventExecutorGroup, String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param baseName the name of the existing handler
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerBefore(EventExecutorGroup group, String baseName,
String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added after an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addAfter(String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param baseName the name of the existing handler
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerAfter(String baseName, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for all connections created by this client. The specified
* handler is added after an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addAfter(EventExecutorGroup, String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param baseName the name of the existing handler
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> addChannelHandlerAfter(EventExecutorGroup group, String baseName,
String name, ChannelHandler handler);
/**
* Removes the {@link ChannelHandler} with the passed {@code name} from the {@link ChannelPipeline} for all
* connections created by this client.
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> removeHandler(String name);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* action to configure all the connections created by the newly created client instance.
*
* @param pipelineConfigurator Action to configure {@link ChannelPipeline}.
*
* @return A new {@link TcpClient} instance.
*/
public abstract <II, OO> TcpClient<II, OO> pipelineConfigurator(Action1<ChannelPipeline> pipelineConfigurator);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* eventLoopGroup for all the connections created by the newly created client instance.
*
* @param eventLoopGroup {@link EventLoopGroup} to use.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> eventLoop(EventLoopGroup eventLoopGroup);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code maxConnections} as the maximum number of concurrent connections created by the newly created client instance.
*
* @param maxConnections Maximum number of concurrent connections to be created by this client.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> maxConnections(int maxConnections);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code idleConnectionsTimeoutMillis} as the time elapsed before an idle connections will be closed by the newly
* created client instance.
*
* @param idleConnectionsTimeoutMillis Time elapsed before an idle connections will be closed by the newly
* created client instance
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withIdleConnectionsTimeoutMillis(long idleConnectionsTimeoutMillis);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code limitDeterminationStrategy} as the strategy to control the maximum concurrent connections created by the
* newly created client instance.
*
* @param limitDeterminationStrategy Strategy to control the maximum concurrent connections created by the
* newly created client instance.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withConnectionPoolLimitStrategy(PoolLimitDeterminationStrategy limitDeterminationStrategy);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code poolIdleCleanupScheduler} for detecting and cleaning idle connections by the newly created client instance.
*
* @param poolIdleCleanupScheduler Scheduled to schedule idle connections cleanup.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withPoolIdleCleanupScheduler(ScheduledExecutorService poolIdleCleanupScheduler);
/**
* Creates a new client instances, inheriting all configurations from this client and disabling idle connection
* cleanup for the newly created client instance.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withNoIdleConnectionCleanup();
/**
* Creates a new client instances, inheriting all configurations from this client and disabling connection
* pooling for the newly created client instance.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withNoConnectionPooling();
/**
* Creates a new client instances, inheriting all configurations from this client and enabling wire logging at the
* passed level for the newly created client instance.
*
* @param wireLoggingLevel Logging level at which the wire logs will be logged. The wire logging will only be done if
* logging is enabled at this level for {@link LoggingHandler}
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> enableWireLogging(LogLevel wireLoggingLevel);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code sslEngineFactory} for all secured connections created by the newly created client instance.
*
* @param sslEngineFactory {@link SSLEngineFactory} for all secured connections created by the newly created client
* instance.
*
* @return A new {@link TcpClient} instance.
*/
public abstract TcpClient<I, O> withSslEngineFactory(SSLEngineFactory sslEngineFactory);
}
Connection Request
This folds ClientConfig
and ChannelPipeline
methods into the request.
Every mutation creates a new instance of the request. In order to do multiple mutations, the cost of creating intermediary request instances can be avoided by using ConnectionRequestUpdater
, created from this ConnectionRequest
using newUpdater()
method.
Every subscription to this request, returns one connection.
/**
* A TCP connection request created via {@link TcpClient#createConnectionRequest()}.
*
* <h2>Mutations</h2>
*
* All mutations to this request that creates a brand new instance.
*
* <h2>Optimizing multiple mutations</h2>
*
* A connection creation may include multiple mutations on a {@link ConnectionRequest}. These mutations will create
* as many objects of {@link ConnectionRequest} and hence create unnecessary garbage. In order to remove this
* memory overhead, this class provides a {@link ConnectionRequestUpdater} which can be obtained via
* {@link #newUpdater()}.
* There is no semantic difference between these two approaches of mutations, this approach, optimizes for lesser
* object creation.
*
* <h2> Inititating connections</h2>
*
* A new connection is initiated every time {@link ConnectionRequest#subscribe()} is called and is the only way of
* creating connections.
*
* @param <I> The type of the objects that are read from this connection.
* @param <O> The type of objects that are written to this connection.
*/
public abstract class ConnectionRequest<I, O> extends Observable<ObservableConnection<I, O>> {
protected ConnectionRequest(OnSubscribe<ObservableConnection<I, O>> f) {
super(f);
}
/**
* Enables read timeout for all the connection created by this request.
*
* @param timeOut Read timeout duration.
* @param timeUnit Read timeout time unit.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract ConnectionRequest<I, O> readTimeOut(int timeOut, TimeUnit timeUnit);
/**
* Creates a new client instances, inheriting all configurations from this client and enabling wire logging at the
* passed level for the newly created client instance.
*
* @param wireLogginLevel Logging level at which the wire logs will be logged. The wire logging will only be done if
* logging is enabled at this level for {@link io.netty.handler.logging.LoggingHandler}
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract ConnectionRequest<I, O> enableWireLogging(LogLevel wireLogginLevel);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* {@code sslEngineFactory} for all secured connections created by the newly created client instance.
*
* @param sslEngineFactory {@link SSLEngineFactory} for all secured connections created by the newly created client
* instance.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract ConnectionRequest<I, O> sslEngineFactory(SSLEngineFactory sslEngineFactory);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by
* this request. The specified handler is added at the first position of the pipeline as specified by
* {@link ChannelPipeline#addFirst(String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerFirst(String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added at the first position of the pipeline as specified by
* {@link ChannelPipeline#addFirst(EventExecutorGroup, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerFirst(EventExecutorGroup group,
String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added at the last position of the pipeline as specified by
* {@link ChannelPipeline#addLast(String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerLast(String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added at the last position of the pipeline as specified by
* {@link ChannelPipeline#addLast(EventExecutorGroup, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerLast(EventExecutorGroup group, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added before an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addBefore(String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param baseName the name of the existing handler
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerBefore(String baseName, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added before an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addBefore(EventExecutorGroup, String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param baseName the name of the existing handler
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerBefore(EventExecutorGroup group,
String baseName,
String name, ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added after an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addAfter(String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param baseName the name of the existing handler
* @param name Name of the handler.
* @param handler Handler instance to add.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerAfter(String baseName, String name,
ChannelHandler handler);
/**
* Adds a {@link ChannelHandler} to {@link ChannelPipeline} for the connections created by this request. The specified
* handler is added after an existing handler with the passed {@code baseName} in the pipeline as specified by
* {@link ChannelPipeline#addAfter(EventExecutorGroup, String, String, ChannelHandler)}
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param group the {@link EventExecutorGroup} which will be used to execute the {@link ChannelHandler}
* methods
* @param baseName the name of the existing handler
* @param name the name of the handler to append
* @param handler the handler to append
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> addChannelHandlerAfter(EventExecutorGroup group,
String baseName,
String name,
ChannelHandler handler);
/**
* Removes the {@link ChannelHandler} with the passed {@code name} from the {@link ChannelPipeline} for all
* connections created by this request.
*
* <em>For better flexibility of pipeline modification, the method {@link #pipelineConfigurator(Action1)} will be more
* convenient.</em>
*
* @param name Name of the handler.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> removeHandler(String name);
/**
* Creates a new client instances, inheriting all configurations from this client and using the passed
* action to configure all the connections created by the newly created client instance.
*
* @param pipelineConfigurator Action to configure {@link ChannelPipeline}.
*
* @return A new instance of the {@link ConnectionRequest} sharing all existing state from this request.
* Use {@link #newUpdater()} if you intend to do multiple mutations to this request, to avoid creating unused
* intermediary {@link ConnectionRequest} objects.
*/
public abstract <II, OO> ConnectionRequest<II, OO> pipelineConfigurator(Action1<ChannelPipeline> pipelineConfigurator);
public abstract ConnectionRequestUpdater<I, O> newUpdater();
}
Examples
Infinite stream read
RxNetty.newTcpClient("localhost", 8080)
.createConnectionRequest()
.flatMap(ObservableConnection::getInput)
.toBlocking()
.forEach(byteBuf -> System.out.println(byteBuf.toString(Charset.defaultCharset())));
Write and Read infinite stream
RxNetty.newTcpClient("localhost", 8080)
.createConnectionRequest()
.flatMap(connection -> connection.writeStringAndFlush("Hello")
.cast(ByteBuf.class)
.concatWith(connection.getInput()))
.toBlocking()
.forEach(byteBuf -> System.out.println(byteBuf.toString(Charset.defaultCharset())));
Add ChannelHandler
RxNetty.newTcpClient("localhost", 8080)
.createConnectionRequest()
.<ByteBuf, String>addChannelHandlerFirst("string-decoder", new SimpleChannelInboundHandler<ByteBuf>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
ctx.fireChannelRead(msg.toString(Charset.defaultCharset()));
}
})
.flatMap(ObservableConnection::getInput)
.toBlocking()
.forEach(System.out::println);
Metadata
Metadata
Assignees
Labels
No labels
Type
Projects
Milestone
Relationships
Development
Issue actions
You can’t perform that action at this time.