io.netty.handler.codec.http.websocketx.WebSocketFrame # isFinalFragment ( ) 源码实例Demo

@Override
protected void decode(final ChannelHandlerContext ctx, final WebSocketFrame msg, final List<Object> out)
    throws Exception {
    if (msg instanceof BinaryWebSocketFrame) {
        ByteBuf content = msg.content();
        // the content is passed to other handlers so they need to be retained.
        content.retain();
        fragments.add(content);
        if (msg.isFinalFragment()) {
            if (fragments.size() == 1) {
                out.add(fragments.get(0));
            } else {
                ByteBuf[] array = fragments.toArray(BYTE_BUF_TYPE);
                out.add(Unpooled.wrappedBuffer(array));
            }
            fragments.clear();
        }
    } else if (msg instanceof TextWebSocketFrame) {
        LOG.warn("Recieved a Websocket text frame. This was not expected. Ignoring it.");
    }
}
 

private void onText(WebSocketFrame frame, String text) throws IOException {
    if (session.isSessionClosed()) {
        //to bad, the channel has already been closed
        //we just ignore messages that are received after we have closed, as the endpoint is no longer in a valid state to deal with them
        //this this should only happen if a message was on the wire when we called close()
        session.close();
        return;
    }

    if (!frame.isFinalFragment()) {
        expectedContinuation = FrameType.TEXT;
    } else {
        expectedContinuation = null;
    }
    final HandlerWrapper handler = getHandler(FrameType.TEXT);
    if (handler != null &&
        (handler.isPartialHandler() || (stringBuffer == null && frame.isFinalFragment()))) {
        invokeTextHandler(text, handler, frame.isFinalFragment());
    } else if (handler != null) {
        if (stringBuffer == null) {
            stringBuffer = new StringBuilder();
        }
        stringBuffer.append(text);
        if (frame.isFinalFragment()) {
            invokeTextHandler(stringBuffer.toString(), handler, frame.isFinalFragment());
            stringBuffer = null;
        }
    }
}
 

private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

        // Check for closing frame
        if (frame instanceof CloseWebSocketFrame) {
        	handleMessageCompleted(ctx, jsonBuffer.toString());
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            SessionRegistry.destroySession(session);
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
        	if (logger.isDebugEnabled())
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
        	jsonBuffer = new StringBuffer();
        	jsonBuffer.append(((TextWebSocketFrame)frame).text());
        }
        else if (frame instanceof ContinuationWebSocketFrame) {
        	if (jsonBuffer != null) {
        		jsonBuffer.append(((ContinuationWebSocketFrame)frame).text());
        	}
        	else {
        		comlog.warn("Continuation frame received without initial frame.");
        	}
        }
        else {
        	throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass()
                    .getName()));
        }

        // Check if Text or Continuation Frame is final fragment and handle if needed.
        if (frame.isFinalFragment()) {
        	handleMessageCompleted(ctx, jsonBuffer.toString());
        	jsonBuffer = null;
        }
    }
 

@Override
protected void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
   if (logger.isTraceEnabled()) {
      logger.trace("Received incoming frame [{}]", frame.getClass().getName());
   }

// Check for closing frame
   if (frame instanceof CloseWebSocketFrame) {
      if (jsonBuffer != null) {
          handleMessageCompleted(ctx, jsonBuffer.toString());
      }
      terminateSocketSessionWithExtremePrejudice(ctx.channel());
      closeWebSocket(ctx, (CloseWebSocketFrame) frame.retain());
      metrics.incSessionDestroyedCounter();
      return;
   }

   if (frame instanceof TextWebSocketFrame) {
      metrics.incFramesReceivedCounter();
      jsonBuffer = new StringBuilder();
      jsonBuffer.append(((TextWebSocketFrame)frame).text());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      metrics.incFramesReceivedCounter();
      if (jsonBuffer != null) {
         jsonBuffer.append(((ContinuationWebSocketFrame)frame).text());
      } else {
         comlog.warn("Continuation frame received without initial frame.");
      }
   } else {
      super.handleWebSocketFrame(ctx, frame);
      return;
   }

   // Check if Text or Continuation Frame is final fragment and handle if needed.
   if (frame.isFinalFragment()) {
    handleMessageCompleted(ctx, jsonBuffer.toString());
    jsonBuffer = null;
   }
}
 

@Override
protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg,
    List<Object> out) throws Exception {
    super.decode(ctx, msg, out);

    if (msg.isFinalFragment()) {
        compressing = false;
    } else if (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame) {
        compressing = true;
    }
}
 

@Override
protected void encode(ChannelHandlerContext ctx, WebSocketFrame msg,
        List<Object> out) throws Exception {
    super.encode(ctx, msg, out);

    if (msg.isFinalFragment()) {
        compressing = false;
    } else if (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame) {
        compressing = true;
    }
}
 

protected void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
   logger.debug("Received incoming frame [{}]", frame.getClass().getName());
   // Check for closing frame
   if (frame instanceof CloseWebSocketFrame) {
      if (frameBuffer != null) {
          handleMessageCompleted(ctx, frameBuffer.toString());
      }
      handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
      return;
   }

   if (frame instanceof PingWebSocketFrame) {
      ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
      return;
   }

   if (frame instanceof PongWebSocketFrame) {
      logger.info("Pong frame received");
      return;
   }

   if (frame instanceof TextWebSocketFrame) {
      frameBuffer = new StringBuilder();
      frameBuffer.append(((TextWebSocketFrame)frame).text());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      if (frameBuffer != null) {
         frameBuffer.append(((ContinuationWebSocketFrame)frame).text());
      } else {
         logger.warn("Continuation frame received without initial frame.");
      }
   } else {
      throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
   }

   // Check if Text or Continuation Frame is final fragment and handle if needed.
   if (frame.isFinalFragment()) {
      handleMessageCompleted(ctx, frameBuffer.toString());
      frameBuffer = null;
   }
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
  if (!frame.isFinalFragment()) {
    LOG.warning("Frame is not final. Chaos may ensue");
  }

  Message message = null;

  if (frame instanceof TextWebSocketFrame) {
    message = new TextMessage(((TextWebSocketFrame) frame).text());
  } else if (frame instanceof BinaryWebSocketFrame) {
    ByteBuf buf = frame.content();
    if (buf.nioBufferCount() != -1) {
      message = new BinaryMessage(buf.nioBuffer());
    } else if (buf.hasArray()) {
      message = new BinaryMessage(ByteBuffer.wrap(buf.array()));
    } else {
      throw new IllegalStateException("Unable to handle bytebuf: " + buf);
    }
  } else if (frame instanceof CloseWebSocketFrame) {
    CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame;
    message = new CloseMessage(closeFrame.statusCode(), closeFrame.reasonText());
  }

  if (message != null) {
    ctx.fireChannelRead(message);
  } else {
    ctx.write(frame);
  }
}
 

@Override
protected void channelRead0(@Nullable ChannelHandlerContext ctx, @Nullable Object msg) throws Exception {
   if (ctx == null || msg == null) {
      return;
   }

   lastPlatformMsg = System.nanoTime();
   Channel ch = ctx.channel();
   if (!handshaker.isHandshakeComplete()) {
      handshaker.finishHandshake(ch, (FullHttpResponse) msg);

      connected = true;
      handshakeFuture.setSuccess();
      return;
   }

   if (msg instanceof FullHttpResponse) {
      log.warn("unxpected full http response: {}", msg);
      ctx.close();
      return;
   }

   WebSocketFrame frame = (WebSocketFrame) msg;
   if (frame instanceof BinaryWebSocketFrame) {
      websocketFrameBuf.clear();
      websocketFrameBuf.writeBytes(frame.content());
   } else if (frame instanceof ContinuationWebSocketFrame){
      if (websocketFrameBuf.isReadable()) {
         websocketFrameBuf.writeBytes(frame.content());
      } else {
         log.warn("continuation frame received without initial frame.");
         ctx.close();
      }
   } else if (frame instanceof PingWebSocketFrame) {
      log.trace("received websocket ping request from platform");
      ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
      lastHubMsg = System.nanoTime();
      return;
   } else if (frame instanceof PongWebSocketFrame) {
      log.trace("received websocket pong response from platform");
      return;
   } else if (frame instanceof CloseWebSocketFrame) {
      log.warn("received websocket close request");
      ctx.close();
      return;
   }

   if (frame.isFinalFragment()) {
      decodeHubFrame(ctx, websocketFrameBuf);
   }
}
 

@Override
protected boolean appendFrameTail(WebSocketFrame msg) {
    return msg.isFinalFragment();
}
 

@Override
protected boolean removeFrameTail(WebSocketFrame msg) {
    return msg.isFinalFragment();
}