下面列出了怎么用 io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler 的API类实例代码及写法,或者点击链接到github查看源代码。
@Override
public void onOpen(Session session, EndpointConfig config) {
String id = session.getId();
log.debug("{}: open ws proxy ", id);
try {
ChannelFuture cf = backend.connect().sync();
Channel channel = cf.channel();
WebSocketClientProtocolHandler wscph = makeWsProtocolHandler(session);
WebSocketClientHandshaker handshaker = wscph.handshaker();
WsHandler handler = new WsHandler(handshaker, channel, session);
channel.pipeline().addLast(new HttpObjectAggregator(1024 * 4),
WebSocketClientCompressionHandler.INSTANCE,
wscph,
handler);
handshaker.handshake(channel);
log.debug("{}: wait messages", id);
session.addMessageHandler(String.class, handler::onFrontString);
session.addMessageHandler(ByteBuffer.class, handler::onFrontBytes);
} catch (Exception e) {
log.error("{}: can not establish ws connect with backed", id, e);
}
}
@SuppressWarnings("FutureReturnValueIgnored")
final void withWebsocketSupport(WebsocketClientSpec websocketClientSpec, boolean compress) {
URI url = websocketUri();
//prevent further header to be sent for handshaking
if (markSentHeaders()) {
// Returned value is deliberately ignored
addHandlerFirst(NettyPipeline.HttpAggregator, new HttpObjectAggregator(8192));
removeHandler(NettyPipeline.HttpMetricsHandler);
if (websocketClientSpec.compress()) {
requestHeaders().remove(HttpHeaderNames.ACCEPT_ENCODING);
// Returned value is deliberately ignored
removeHandler(NettyPipeline.HttpDecompressor);
// Returned value is deliberately ignored
addHandlerFirst(NettyPipeline.WsCompressionHandler, WebSocketClientCompressionHandler.INSTANCE);
}
if (log.isDebugEnabled()) {
log.debug(format(channel(), "Attempting to perform websocket handshake with {}"), url);
}
WebsocketClientOperations ops = new WebsocketClientOperations(url, websocketClientSpec, this);
if (!rebind(ops)) {
log.error(format(channel(), "Error while rebinding websocket in channel attribute: " +
get(channel()) + " to " + ops));
}
}
}
/**
* @return true if the handshake is done properly.
* @throws URISyntaxException throws if there is an error in the URI syntax.
* @throws InterruptedException throws if the connecting the server is interrupted.
*/
public boolean handhshake() throws InterruptedException, URISyntaxException, SSLException, ProtocolException {
boolean isSuccess;
URI uri = new URI(url);
String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
final int port;
if (uri.getPort() == -1) {
if ("ws".equalsIgnoreCase(scheme)) {
port = 80;
} else if ("wss".equalsIgnoreCase(scheme)) {
port = 443;
} else {
port = -1;
}
} else {
port = uri.getPort();
}
if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
logger.error("Only WS(S) is supported.");
return false;
}
final boolean ssl = "wss".equalsIgnoreCase(scheme);
final SslContext sslCtx;
if (ssl) {
sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
group = new NioEventLoopGroup();
HttpHeaders headers = new DefaultHttpHeaders();
for (Map.Entry<String, String> entry : customHeaders.entrySet()) {
headers.add(entry.getKey(), entry.getValue());
}
// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
handler = new WebSocketClientHandler(
WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, subProtocol, true, headers),
latch);
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
}
p.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192),
WebSocketClientCompressionHandler.INSTANCE, handler);
}
});
channel = bootstrap.connect(uri.getHost(), port).sync().channel();
isSuccess = handler.handshakeFuture().sync().isSuccess();
logger.info("WebSocket Handshake successful : " + isSuccess);
return isSuccess;
}
/**
* @return true if the handshake is done properly.
* @throws URISyntaxException throws if there is an error in the URI syntax.
* @throws InterruptedException throws if the connecting the server is interrupted.
*/
public boolean handhshake() throws InterruptedException, URISyntaxException, SSLException, ProtocolException {
boolean isSuccess;
URI uri = new URI(url);
String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
final int port;
if (uri.getPort() == -1) {
if ("ws".equalsIgnoreCase(scheme)) {
port = 80;
} else if ("wss".equalsIgnoreCase(scheme)) {
port = 443;
} else {
port = -1;
}
} else {
port = uri.getPort();
}
if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
logger.error("Only WS(S) is supported.");
return false;
}
final boolean ssl = "wss".equalsIgnoreCase(scheme);
final SslContext sslCtx;
if (ssl) {
sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
group = new NioEventLoopGroup();
HttpHeaders headers = new DefaultHttpHeaders();
for (Map.Entry<String, String> entry : customHeaders.entrySet()) {
headers.add(entry.getKey(), entry.getValue());
}
// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
handler = new WebSocketClientHandler(
WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, subProtocol, true, headers),
latch);
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
}
p.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192),
WebSocketClientCompressionHandler.INSTANCE, handler);
}
});
channel = bootstrap.connect(uri.getHost(), port).sync().channel();
isSuccess = handler.handshakeFuture().sync().isSuccess();
logger.info("WebSocket Handshake successful : " + isSuccess);
return isSuccess;
}
/**
* @return true if the handshake is done properly.
* @throws URISyntaxException throws if there is an error in the URI syntax.
* @throws InterruptedException throws if the connecting the server is interrupted.
*/
public boolean handhshake() throws InterruptedException, URISyntaxException, SSLException {
boolean isDone;
URI uri = new URI(url);
String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
final int port;
if (uri.getPort() == -1) {
if ("ws".equalsIgnoreCase(scheme)) {
port = 80;
} else if ("wss".equalsIgnoreCase(scheme)) {
port = 443;
} else {
port = -1;
}
} else {
port = uri.getPort();
}
if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
logger.error("Only WS(S) is supported.");
return false;
}
final boolean ssl = "wss".equalsIgnoreCase(scheme);
final SslContext sslCtx;
if (ssl) {
sslCtx = SslContextBuilder.forClient()
.trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
group = new NioEventLoopGroup();
HttpHeaders headers = new DefaultHttpHeaders();
customHeaders.entrySet().forEach(
header -> headers.add(header.getKey(), header.getValue())
);
try {
// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
handler =
new WebSocketClientHandler(
WebSocketClientHandshakerFactory.newHandshaker(
uri, WebSocketVersion.V13, subProtocol,
true, headers));
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
}
p.addLast(
new HttpClientCodec(),
new HttpObjectAggregator(8192),
WebSocketClientCompressionHandler.INSTANCE,
handler);
}
});
channel = b.connect(uri.getHost(), port).sync().channel();
isDone = handler.handshakeFuture().sync().isSuccess();
logger.debug("WebSocket Handshake successful : " + isDone);
return isDone;
} catch (Exception e) {
logger.error("Handshake unsuccessful : " + e.getMessage(), e);
return false;
}
}