Wie verwende ich FixedChannelPool mit WebSockets?

Question

Hallo Kolleginnen und Entwickler ...

Ich versuche WebSocket feste Verbindung Pool zu implementieren, leider netty schlechte Führung vorgesehen für FixedChannelPool und Maximum „was man bekommt“ ist ihr Unit Tests

Client-Code, den ich verwende Source Code Beispiel

und von this post fand heraus, dass bootstrap.handler(new ChannelInitializer<>()) von ChannelPool

außer Kraft gesetzt wird Dann habe ich versucht 0.123.867 zu bewegenBlock:

public class SomeConnectionPoolHandler implements ChannelPoolHandler { 

    private final URI uri = URI.create("ws://some.url:80"); 

    @Override 
    public void channelCreated(Channel ch) { 
     String protocol = uri.getScheme(); 
     if (!"ws".equals(protocol)) { 
      throw new IllegalArgumentException("Unsupported protocol: " + protocol); 
     } 

     DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders(); 
     final WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, httpHeaders, 1280000); 
     final WebSocketClientHandler handler = new WebSocketClientHandler(handshaker, connection); 
     ChannelPipeline pipeline = ch.pipeline(); 
     pipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(65536), handler); 
    } 

    @Override 
    public void channelReleased(Channel ch) { 
     //TODO 
    } 

    @Override 
    public void channelAcquired(Channel ch) { 
     //TODO 
    } 
} 

Und jetzt mein Client-Code wie folgt aussieht:

public class Application { 
    private final URI uri = URI.create("ws://some.url:80"); 
    public static void main(String[] args) { 
     Channel ch = null; 
     try { 
      String protocol = uri.getScheme(); 
      if (!"ws".equals(protocol)) { 
       throw new IllegalArgumentException("Unsupported protocol: " + protocol); 
      } 
      Bootstrap b = new Bootstrap(); 
      b.option(ChannelOption.SO_KEEPALIVE, true); 
      b.remoteAddress(uri.getHost(), uri.getPort()); 

      b.group(group); 
      b.channel(NioSocketChannel.class); 
      SomeConnectionPoolHandler connectionPoolHandler = new SomeConnectionPoolHandler(); 

      channelPool = new FixedChannelPool(b, connectionPoolHandler, 8); 
      ch = channelPool.acquire().sync().get(); 
      ch.writeAndFlush(new TextWebSocketFrame("test")).sync(); 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } finally { 
      if(ch != null) { 
       channelPool.release(ch); 
      } 
     } 
    } 
} 

Jetzt erhalte ich Fehler:

java.lang.UnsupportedOperationException: unsupported message type: TextWebSocketFrame (expected: ByteBuf, FileRegion) 
    at io.netty.channel.nio.AbstractNioByteChannel.filterOutboundMessage(AbstractNioByteChannel.java:266) 
    at io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:799) 
    at io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1291) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:738) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:730) 
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:816) 
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:723) 
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.write(CombinedChannelDuplexHandler.java:528) 
    at io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:101) 
    at io.netty.channel.CombinedChannelDuplexHandler.write(CombinedChannelDuplexHandler.java:348) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:738) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:730) 
    at io.netty.channel.AbstractChannelHandlerContext.access$1900(AbstractChannelHandlerContext.java:38) 
    at io.netty.channel.AbstractChannelHandlerContext$AbstractWriteTask.write(AbstractChannelHandlerContext.java:1089) 
    at io.netty.channel.AbstractChannelHandlerContext$WriteAndFlushTask.write(AbstractChannelHandlerContext.java:1136) 
    at io.netty.channel.AbstractChannelHandlerContext$AbstractWriteTask.run(AbstractChannelHandlerContext.java:1078) 
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) 
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403) 
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:442) 
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858) 
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144) 
    at java.lang.Thread.run(Thread.java:745) 

Answer 1

Ok, ich Lösung für mein Problem zu finden.

Das Problem ist, dass ich versuchte, Daten in Channel zu schreiben, bevor WebSocket Handshake abgeschlossen ist.

In handler source code können Sie sehen, dass es Methode hat, ChannelFuture für WebSocket Handshake zu erhalten.

public ChannelFuture handshakeFuture() { 
     return handshakeFuture; 
    } 

Jetzt, nach Channel von FixedChannelPool Erwerb, ich rufe Methode (gibt zurück, ob Handshake abgeschlossen wurde oder nicht):

private boolean isConnectionReady(Channel ch) { 
     try { 
      WebSocketClientHandler handler = (WebSocketClientHandler) ch.pipeline().get("handler"); 
      return handler.handshakeFuture().sync().isSuccess(); 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } 
     return false; 
    } 

Auch hinzugefügt Name für WebSocketClientHandler in channelCreated(Channel ch)

DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders(); 
final WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, httpHeaders, 1280000); 
final WebSocketClientHandler handler = new WebSocketClientHandler(handshaker, connection); 
ChannelPipeline pipeline = ch.pipeline(); 
pipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(65536)); 
pipeline.addLast("handler", handler); 

Ich weiß, dass dieser Code jetzt schrecklich aussieht und überarbeitet werden sollte.