java类org.jboss.netty.channel.Channel的实例源码

@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel,
        ChannelBuffer buffer) throws Exception {
    if (!channel.isConnected()) {
        // In testing, I see decode being called AFTER decode last.
        // This check avoids that from reading corrupted frames
        return null;
    }

    // Note that a single call to decode results in reading a single
    // OFMessage from the channel buffer, which is passed on to, and processed
    // by, the controller (in OFChannelHandler).
    // This is different from earlier behavior (with the original openflowj),
    // where we parsed all the messages in the buffer, before passing on
    // a list of the parsed messages to the controller.
    // The performance *may or may not* not be as good as before.
    OFMessageReader<OFMessage> reader = OFFactories.getGenericReader();
    OFMessage message = reader.readFrom(buffer);

    return message;
}

/**
 * Handle a generic {@link SyncMessage} and dispatch to an appropriate
 * handler
 * @param bsm the message
 * @param channel the channel on which the message arrived
 */
protected void handleSyncMessage(SyncMessage bsm, Channel channel) {
    switch (channelState) {
        case OPEN:
        case CONNECTED:
            switch (bsm.getType()) {
                case HELLO:
                    handshake(bsm.getHello(), channel);
                    break;
                case ECHO_REQUEST:
                    handleEchoRequest(bsm.getEchoRequest(), channel);
                    break;
                case ERROR:
                    handleError(bsm.getError(), channel);
                    break;
                default:
                    // ignore
            }
            break;
        case AUTHENTICATED:
            handleSMAuthenticated(bsm, channel);
            break;
    }
}

@Override
protected Object decode(
        ChannelHandlerContext ctx, Channel channel, ChannelBuffer buf) throws Exception {

    if (buf.readableBytes() >= MESSAGE_HEADER) {
        int length = Integer.parseInt(buf.toString(2, 2, StandardCharsets.US_ASCII)) + 5;
        if (buf.readableBytes() >= length) {
            ChannelBuffer frame = buf.readBytes(length);
            while (buf.readable() && buf.getUnsignedByte(buf.readerIndex()) != '$') {
                buf.readByte();
            }
            return frame;
        }
    }

    return null;
}

/**
 * Tests adjOk() method.
 */
@Test
public void testAdjOk() throws Exception {
    channel = EasyMock.createMock(Channel.class);
    ospfInterface.setInterfaceType(OspfInterfaceType.BROADCAST.value());
    ospfInterface.setIpAddress(Ip4Address.valueOf("2.2.2.2"));
    ospfNbr1 = new OspfNbrImpl(ospfArea, ospfInterface, Ip4Address.valueOf("1.1.1.1"),
                               Ip4Address.valueOf("2.2.2.2"), 2,
                               topologyForDeviceAndLink);
    ospfNbr1.setState(OspfNeighborState.TWOWAY);
    ospfNbr1.setNeighborDr(Ip4Address.valueOf("2.2.2.2"));
    ospfNbr1.adjOk(channel);
    assertThat(ospfNbr1, is(notNullValue()));

    ospfInterface.setInterfaceType(OspfInterfaceType.POINT_TO_POINT.value());
    ospfNbr1 = new OspfNbrImpl(ospfArea, ospfInterface, Ip4Address.valueOf("1.1.1.1"),
                               Ip4Address.valueOf("2.2.2.2"), 2,
                               topologyForDeviceAndLink);
    channel = null;
    channel = EasyMock.createMock(Channel.class);
    ospfNbr1.adjOk(channel);
    assertThat(ospfNbr1, is(notNullValue()));
}

@Test(timeout = 60000)
public void testCommit() throws Exception {
  HdfsFileStatus status = nn.getRpcServer().getFileInfo("/tmp/bar");
  long dirId = status.getFileId();
  FileHandle handle = new FileHandle(dirId);
  XDR xdr_req = new XDR();
  COMMIT3Request req = new COMMIT3Request(handle, 0, 5);
  req.serialize(xdr_req);

  Channel ch = Mockito.mock(Channel.class);

  // Attempt by an unpriviledged user should fail.
  COMMIT3Response response1 = nfsd.commit(xdr_req.asReadOnlyWrap(),
      ch, 1, securityHandlerUnpriviledged,
      new InetSocketAddress("localhost", 1234));
  assertEquals("Incorrect return code:", Nfs3Status.NFS3ERR_ACCES,
      response1.getStatus());

  // Attempt by a priviledged user should pass.
  COMMIT3Response response2 = nfsd.commit(xdr_req.asReadOnlyWrap(),
      ch, 1, securityHandler,
      new InetSocketAddress("localhost", 1234));
  assertEquals("Incorrect COMMIT3Response:", null, response2);
}

private WriteCtx checkRepeatedWriteRequest(WRITE3Request request,
    Channel channel, int xid) {
  OffsetRange range = new OffsetRange(request.getOffset(),
      request.getOffset() + request.getCount());
  WriteCtx writeCtx = pendingWrites.get(range);
  if (writeCtx== null) {
    return null;
  } else {
    if (xid != writeCtx.getXid()) {
      LOG.warn("Got a repeated request, same range, with a different xid: "
          + xid + " xid in old request: " + writeCtx.getXid());
      //TODO: better handling.
    }
    return writeCtx;  
  }
}

@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buf) throws Exception {

    // Check minimum length
    if (buf.readableBytes() < MESSAGE_MINIMUM_LENGTH) {
        return null;
    }

    // Check for sync packet
    if (buf.getUnsignedShort(buf.readerIndex()) == 0xFAF8) {
        ChannelBuffer syncMessage = buf.readBytes(8);
        if (channel != null) {
            channel.write(syncMessage);
        }
    }

    return super.decode(ctx, channel, buf);
}

private void sendPhotoRequest(Channel channel, int pictureId) {
    if (channel != null) {
        ChannelBuffer photo = photos.get(pictureId);
        ChannelBuffer response = ChannelBuffers.dynamicBuffer();
        response.writeShort(0x7878); // header
        response.writeByte(15); // size
        response.writeByte(MSG_X1_PHOTO_DATA);
        response.writeInt(pictureId);
        response.writeInt(photo.writerIndex());
        response.writeShort(Math.min(photo.writableBytes(), 1024));
        response.writeShort(++serverIndex);
        response.writeShort(Checksum.crc16(Checksum.CRC16_X25,
                response.toByteBuffer(2, response.writerIndex() - 2)));
        response.writeByte('\r'); response.writeByte('\n'); // ending
        channel.write(response);
    }
}

@Override
protected void handleRegisterRequest(RegisterRequestMessage request,
                                     Channel channel) {
    try {
        Scope scope = TProtocolUtil.getScope(request.store.getScope());
        if (request.store.isPersist())
            syncManager.registerPersistentStore(request.store.storeName,
                                                scope);
        else
            syncManager.registerStore(request.store.storeName, scope);
        RegisterResponseMessage m = new RegisterResponseMessage();
        AsyncMessageHeader header = new AsyncMessageHeader();
        header.setTransactionId(request.getHeader().getTransactionId());
        m.setHeader(header);
        SyncMessage bsm =
                new SyncMessage(MessageType.REGISTER_RESPONSE);
        bsm.setRegisterResponse(m);
        channel.write(bsm);
    } catch (Exception e) {
        channel.write(getError(request.getHeader().getTransactionId(), e,
                               MessageType.REGISTER_REQUEST));
    }
}

@Override
protected Object encode(ChannelHandlerContext ctx, Channel channel,
                        Object message) throws Exception {
    if (message instanceof SyncMessage) {
        ChannelBuffer buf = new DynamicChannelBuffer(512);
        ChannelBufferOutputStream os = new ChannelBufferOutputStream(buf);
        TCompactProtocol thriftProtocol =
                new TCompactProtocol(new TIOStreamTransport(os));
        ((SyncMessage) message).write(thriftProtocol);

        ChannelBuffer len = ChannelBuffers.buffer(4);
        len.writeInt(buf.readableBytes());
        return ChannelBuffers.wrappedBuffer(len, buf);
    }
    return message;
}

/**
 * At this point, the router has sent and received an entire sequence of DD packets.
 * Now it must be determined whether the new state is FULL, or LS Request packets
 * have to be send.
 *
 * @param message OSPF message instance
 * @param ch      netty channel handler
 */
public void exchangeDone(OspfMessage message, Channel ch) {
    log.debug("OSPFNbr::exchangeDone...!!!");
    stopRxMtDdTimer();

    OspfPacketHeader header = (OspfPacketHeader) message;

    if (state == OspfNeighborState.EXCHANGE) {
        if (lsReqList.isEmpty()) {
            state = OspfNeighborState.FULL;
            //handler.addDeviceInformation(this);
            //handler.addLinkInformation(this, topLevelTlvs);
        } else {
            state = OspfNeighborState.LOADING;
            LsRequest lsRequest = buildLsRequest();
            //Setting the destination address
            lsRequest.setDestinationIp(header.sourceIp());
            byte[] messageToWrite = getMessage(lsRequest);
            ch.write(messageToWrite);

            setLastSentLsrPacket(lsRequest);
            startRxMtLsrTimer(ch);
        }
    }
}

private Position parseCommandResponse(Channel channel, SocketAddress remoteAddress, ChannelBuffer buf) {
    DeviceSession deviceSession = getDeviceSession(channel, remoteAddress);
    if (deviceSession == null) {
        return null;
    }

    Position position = new Position();
    position.setProtocol(getProtocolName());
    position.setDeviceId(deviceSession.getDeviceId());

    getLastLocation(position, null);

    int responseTextLength = buf.bytesBefore((byte) 0);
    if (responseTextLength < 0) {
        responseTextLength = CMD_RESPONSE_SIZE - 3;
    }
    position.set(Position.KEY_RESULT, buf.readBytes(responseTextLength).toString(StandardCharsets.UTF_8));

    return position;
}

public boolean hasChannel(Channel channel) {
    for (Channel channel1 : allChannels) {
        if (channel.equals(channel1)) {
            return true;
        }
    }
    return false;
}

/**
 * Tests oneWayReceived() method.
 */
@Test
public void testOneWayReceived() throws Exception {
    ospfMessage = new HelloPacket();
    ospfNbr.setState(OspfNeighborState.ATTEMPT);
    channel = EasyMock.createMock(Channel.class);
    ospfNbr.oneWayReceived(ospfMessage, channel);
    channel1 = EasyMock.createMock(Channel.class);
    ospfNbr.setState(OspfNeighborState.DOWN);
    ospfNbr.oneWayReceived(ospfMessage, channel1);
    channel2 = EasyMock.createMock(Channel.class);
    ospfNbr.setState(OspfNeighborState.TWOWAY);
    ospfNbr.oneWayReceived(ospfMessage, channel2);
    assertThat(ospfNbr, is(notNullValue()));
}

@Override
public void execute(Channel chan, String[] command) throws DatastoreException
{
    m_counter.incrementAndGet();
    if (chan.isConnected())
    {
        Package thisPackage = getClass().getPackage();
        String versionString = thisPackage.getImplementationTitle()+" "+thisPackage.getImplementationVersion();
        chan.write(versionString+"\n");
    }
}

/**
 * 尝试连接
 */
public void connect() {
       ChannelFuture future = bootstrap.connect(inetAddr);
       try{
           boolean ret = future.awaitUninterruptibly(Constants.TIMEOUT_CONNECTION_MILLSECOND, TimeUnit.MILLISECONDS);
           if (ret && future.isSuccess()) {
               Channel newChannel = future.getChannel();
               newChannel.setInterestOps(Channel.OP_READ_WRITE);
               try {
                   // 关闭旧的连接
                   Channel oldChannel = NettyRpcConnection.this.channel;
                   if (oldChannel != null) {
                       logger.info("Close old netty channel {} on create new netty channel {}", oldChannel, newChannel);
                       oldChannel.close();
                   }
               } finally {
                   if (!isConnected()) {
                       try {
                           logger.info("Close new netty channel {}, because the client closed.", newChannel);
                           newChannel.close();
                       } finally {
                        NettyRpcConnection.this.channel = null;
                       }
                   } else {
                    NettyRpcConnection.this.channel = newChannel;
                   }
               }
           } else if (null != future.getCause()) {
            logger.error("connect fail", future.getCause());
            throw new RuntimeException("connect error", future.getCause());
           } else {
            logger.error("connect fail,connstr: "+this.getConnStr());
            throw new RuntimeException("connect error");
           }
       }finally{
           if (! isConnected()) {
               future.cancel();
           }
       }
}

/**
 * Starts Ls request retransmission executor task.
 *
 * @param ch Netty channel instance
 */
private void startRxMtLsrTimer(Channel ch) {
    if (!rxmtLsrTimerScheduled) {
        log.debug("OSPFNbr::startRxMtLsrTimer...!!!");
        long retransmitIntrvl = ospfInterface.reTransmitInterval();
        rxmtLsrPacketTask = new InternalRxmtLsrPacket(ch);
        exServiceRxmtLsr = Executors.newSingleThreadScheduledExecutor();
        exServiceRxmtLsr.scheduleAtFixedRate(rxmtLsrPacketTask, retransmitIntrvl,
                                             retransmitIntrvl, TimeUnit.SECONDS);
        rxmtLsrTimerScheduled = true;
    }
}

@Override
protected com.alibaba.dubbo.remoting.Channel getChannel() {
    Channel c = channel;
    if (c == null || ! c.isConnected())
        return null;
    return NettyChannel.getOrAddChannel(c, getUrl(), this);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
    throws Exception {
  Channel ch = e.getChannel();
  Throwable cause = e.getCause();

  if (LOG.isDebugEnabled())
    LOG.debug(cause.getMessage());
  ch.close().addListener(ChannelFutureListener.CLOSE);
}

@Override
protected void handleHello(HelloMessage hello, Channel channel) {
    syncManager.remoteNodeId = hello.getNodeId();
    syncManager.ready = true;
    synchronized (syncManager.readyNotify) {
        syncManager.notifyAll();
    }
}

private Object decodeOther(Channel channel, SocketAddress remoteAddress, String sentence, String type) {
    Parser parser = new Parser(PATTERN, sentence);
    Position position = initPosition(parser, channel, remoteAddress);
    if (position == null) {
        return null;
    }

    int reportType = parser.nextInt(0);
    if (type.equals("NMR")) {
        position.set(Position.KEY_MOTION, reportType == 1);
    } else if (type.equals("SOS")) {
        position.set(Position.KEY_ALARM, Position.ALARM_SOS);
    }

    decodeLocation(position, parser);

    position.set(Position.KEY_ODOMETER, parser.nextDouble(0) * 1000);
    position.set(Position.KEY_BATTERY_LEVEL, parser.nextInt(0));

    position.set(Position.KEY_ODOMETER, parser.nextDouble(0) * 1000);

    decodeDeviceTime(position, parser);

    if (Context.getConfig().getBoolean(getProtocolName() + ".ack") && channel != null) {
        channel.write("+SACK:" + parser.next() + "$", remoteAddress);
    }

    return position;
}

@Override
protected void handleCursorResponse(CursorResponseMessage response,
                                    Channel channel) {
    SyncReply reply = new SyncReply(null, response.getValues(), true, 
                                    null, response.getCursorId());
    syncManager.dispatchReply(response.getHeader().getTransactionId(), 
                              reply);
}

@Override
protected Object decode(
        ChannelHandlerContext ctx,
        Channel channel,
        ChannelBuffer buf) throws Exception {

    if (buf.readableBytes() < 2) {
        return null;
    }

    int index = buf.indexOf(buf.readerIndex() + 1, buf.writerIndex(), (byte) 0x04);
    if (index != -1) {
        ChannelBuffer result = ChannelBuffers.buffer(ByteOrder.LITTLE_ENDIAN, index + 1 - buf.readerIndex());

        while (buf.readerIndex() <= index) {
            int b = buf.readUnsignedByte();
            if (b == 0x10) {
                result.writeByte(buf.readUnsignedByte() - 0x20);
            } else {
                result.writeByte(b);
            }
        }

        return result;
    }

    return null;
}

/**
 * Sends LS PDU message to channel.
 *
 * @param lsp     LS PDU message instance
 * @param channel channel instance
 */
private void sendLsp(LsPdu lsp, Channel channel) {
    byte[] lspBytes = lsp.asBytes();
    lspBytes = IsisUtil.addLengthAndMarkItInReserved(lspBytes, IsisConstants.LENGTHPOSITION,
                                                     IsisConstants.LENGTHPOSITION + 1,
                                                     IsisConstants.RESERVEDPOSITION);
    lspBytes = IsisUtil.addChecksum(lspBytes, IsisConstants.CHECKSUMPOSITION,
                                    IsisConstants.CHECKSUMPOSITION + 1);
    //write to the channel
    if (channel != null && channel.isConnected() && channel.isOpen()) {
        channel.write(IsisUtil.framePacket(lspBytes, interfaceIndex));
    }
}

public NettyTransportChannel(NettyTransport transport, TransportServiceAdapter transportServiceAdapter, String action, Channel channel, long requestId, Version version, String profileName) {
    this.transportServiceAdapter = transportServiceAdapter;
    this.version = version;
    this.transport = transport;
    this.action = action;
    this.channel = channel;
    this.requestId = requestId;
    this.profileName = profileName;
}

/**
 * Sends the partial sequence number PDU.
 *
 * @param lspEntryRequestList list of lsp entry request
 * @param isisPduType         intermediate system PDU type
 * @param channel             netty channel instance
 */
private void sendPsnPduMessage(List<LspEntry> lspEntryRequestList, IsisPduType isisPduType, Channel channel) {
    IsisHeader isisHeader = new LspGenerator().getHeader(isisPduType);
    Psnp psnp = new Psnp(isisHeader);
    psnp.setSourceId(lspKeyP2P(this.systemId));
    TlvHeader tlvHeader = new TlvHeader();
    tlvHeader.setTlvType(TlvType.LSPENTRY.value());
    tlvHeader.setTlvLength(0);
    LspEntriesTlv lspEntriesTlv = new LspEntriesTlv(tlvHeader);
    for (LspEntry lspEntry : lspEntryRequestList) {
        lspEntry.setLspChecksum(0);
        lspEntry.setLspSequenceNumber(0);
        lspEntry.setRemainingTime(0);
        lspEntriesTlv.addLspEntry(lspEntry);
    }
    psnp.addTlv(lspEntriesTlv);
    //write it to channel buffer.
    byte[] psnpBytes = psnp.asBytes();
    psnpBytes = IsisUtil.addLengthAndMarkItInReserved(psnpBytes, IsisConstants.LENGTHPOSITION,
                                                      IsisConstants.LENGTHPOSITION + 1,
                                                      IsisConstants.RESERVEDPOSITION);
    flagValue = false;
    //write to the channel
    if (channel != null && channel.isConnected() && channel.isOpen()) {
        channel.write(IsisUtil.framePacket(psnpBytes, interfaceIndex));
    }
}

private void sendResponse(Channel channel, HttpResponseStatus status) {
    if (channel != null) {
        HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
        response.headers().add(HttpHeaders.Names.CONTENT_LENGTH, 0);
        channel.write(response);
    }
}

@Override
protected Object decode(
        ChannelHandlerContext ctx, Channel channel, ChannelBuffer buf) throws Exception {

    if (buf.readableBytes() < 10) {
        return null;
    }

    int beginIndex = buf.indexOf(buf.readerIndex(), buf.writerIndex(), new StringFinder("$$"));
    if (beginIndex == -1) {
        return null;
    } else if (beginIndex > buf.readerIndex()) {
        buf.readerIndex(beginIndex);
    }

    int length;

    int flagIndex = buf.indexOf(buf.readerIndex(), buf.writerIndex(), new StringFinder("AA"));
    if (flagIndex != -1 && flagIndex - beginIndex == 6) {
        length = Integer.parseInt(buf.toString(buf.readerIndex() + 2, 4, StandardCharsets.US_ASCII));
    } else {
        length = Integer.parseInt(buf.toString(buf.readerIndex() + 2, 2, StandardCharsets.US_ASCII), 16);
    }

    if (length <= buf.readableBytes()) {
        return buf.readBytes(length);
    }

    return null;
}

@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
    log.debug("MESSAGE IS RECEIVED.");
    if (!channel.isConnected()) {
        log.info("Channel is not connected.");
        return null;
    }

    HexDump.dump(buffer);

    BgpMessageReader<BgpMessage> reader = BgpFactories.getGenericReader();
    List<BgpMessage> msgList = (List<BgpMessage>) ctx.getAttachment();

    if (msgList == null) {
        msgList = new LinkedList<>();
    }

    try {
        while (buffer.readableBytes() > 0) {
            buffer.markReaderIndex();
            BgpHeader bgpHeader = new BgpHeader();
            BgpMessage message = reader.readFrom(buffer, bgpHeader);
            msgList.add(message);
        }

        return msgList;
    } catch (Exception e) {
        log.debug("Bgp protocol message decode error");
        buffer.resetReaderIndex();
        buffer.discardReadBytes();
        ctx.setAttachment(msgList);
    }
    return null;
}

@Override
protected Object encode(ChannelHandlerContext ctx, Channel ch, Object msg) throws Exception {
    com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer =
        com.alibaba.dubbo.remoting.buffer.ChannelBuffers.dynamicBuffer(1024);
    NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
    try {
        codec.encode(channel, buffer, msg);
    } finally {
        NettyChannel.removeChannelIfDisconnected(ch);
    }
    return ChannelBuffers.wrappedBuffer(buffer.toByteBuffer());
}