io.netty.util.internal.PlatformDependent#throwException ( )源码实例Demo

@Override
protected void startAfter(ChannelFuture future){
    //Exception thrown
    Throwable cause = future.cause();
    if(cause != null){
        PlatformDependent.throwException(cause);
    }

    logger.info("{} start (version = {}, port = {}, pid = {}, protocol = {}, os = {}) ...",
            getName(),
            ServerInfo.getServerNumber(),
            getPort()+"",
            HostUtil.getPid()+"",
            protocolHandlers,
            HostUtil.getOsName()
            );
}
 

private void addPseudoHeader(CharSequence csName, CharSequence csValue) {
  AsciiString name = requireAsciiString(csName);
  AsciiString value = requireAsciiString(csValue);

  if (equals(PATH_HEADER, name)) {
    path = value;
  } else if (equals(AUTHORITY_HEADER, name)) {
    authority = value;
  } else if (equals(METHOD_HEADER, name)) {
    method = value;
  } else if (equals(SCHEME_HEADER, name)) {
    scheme = value;
  } else {
    PlatformDependent.throwException(
        connectionError(PROTOCOL_ERROR, "Illegal pseudo-header '%s' in request.", name));
  }
}
 

@Override
public void goAwayReceived(final int lastKnownStream, long errorCode, ByteBuf debugData) {
    localEndpoint.lastStreamKnownByPeer(lastKnownStream);
    for (int i = 0; i < listeners.size(); ++i) {
        try {
            listeners.get(i).onGoAwayReceived(lastKnownStream, errorCode, debugData);
        } catch (Throwable cause) {
            logger.error("Caught Throwable from listener onGoAwayReceived.", cause);
        }
    }

    try {
        forEachActiveStream(new Http2StreamVisitor() {
            @Override
            public boolean visit(Http2Stream stream) {
                if (stream.id() > lastKnownStream && localEndpoint.isValidStreamId(stream.id())) {
                    stream.close();
                }
                return true;
            }
        });
    } catch (Http2Exception e) {
        PlatformDependent.throwException(e);
    }
}
 

private void handleUnwrapThrowable(ChannelHandlerContext ctx, Throwable cause) {
    try {
        // We should attempt to notify the handshake failure before writing any pending data. If we are in unwrap
        // and failed during the handshake process, and we attempt to wrap, then promises will fail, and if
        // listeners immediately close the Channel then we may end up firing the handshake event after the Channel
        // has been closed.
        if (handshakePromise.tryFailure(cause)) {
            ctx.fireUserEventTriggered(new SslHandshakeCompletionEvent(cause));
        }

        // We need to flush one time as there may be an alert that we should send to the remote peer because
        // of the SSLException reported here.
        wrapAndFlush(ctx);
    } catch (SSLException ex) {
        logger.debug("SSLException during trying to call SSLEngine.wrap(...)" +
                " because of an previous SSLException, ignoring...", ex);
    } finally {
        // ensure we always flush and close the channel.
        setHandshakeFailure(ctx, cause, true, false, true);
    }
    PlatformDependent.throwException(cause);
}
 

private int forEachByteDesc0(int index, int length, ByteBufProcessor processor) {

        if (processor == null) {
            throw new NullPointerException("processor");
        }

        if (length == 0) {
            return -1;
        }

        int i = index + length - 1;
        try {
            do {
                if (processor.process(_getByte(i))) {
                    i --;
                } else {
                    return i;
                }
            } while (i >= index);
        } catch (Exception e) {
            PlatformDependent.throwException(e);
        }

        return -1;
    }
 

private void runFinishPeerReadTask(final VirtualChannel peer) {
    // If the peer is writing, we must wait until after reads are completed for that peer before we can read. So
    // we keep track of the task, and coordinate later that our read can't happen until the peer is done.
    final Runnable finishPeerReadTask = new Runnable() {
        @Override
        public void run() {
            finishPeerRead0(peer);
        }
    };
    try {
        if (peer.writeInProgress) {
            peer.finishReadFuture = peer.eventLoop().submit(finishPeerReadTask);
        } else {
            peer.eventLoop().execute(finishPeerReadTask);
        }
    } catch (Throwable cause) {
        close();
        peer.close();
        PlatformDependent.throwException(cause);
    }
}
 

private V initialize(InternalThreadLocalMap threadLocalMap) {
    V v = null;
    try {
        v = initialValue();
    } catch (Exception e) {
        PlatformDependent.throwException(e);
    }

    threadLocalMap.setIndexedVariable(index, v);
    addToVariablesToRemove(threadLocalMap, this);
    return v;
}
 

@Override
public long convertToTimeMillis(CharSequence value) {
    Date date = DateFormatter.parseHttpDate(value);
    if (date == null) {
        PlatformDependent.throwException(new ParseException("header can't be parsed into a Date: " + value, 0));
        return 0;
    }
    return date.getTime();
}
 

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    DatagramChannel ch = javaChannel();
    ChannelConfig config = config();
    RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();

    ByteBuf data = allocHandle.allocate(config.getAllocator());
    allocHandle.attemptedBytesRead(data.writableBytes());
    boolean free = true;
    try {
        ByteBuffer nioData = data.internalNioBuffer(data.writerIndex(), data.writableBytes());
        int pos = nioData.position();
        int read = ch.read(nioData);
        if (read <= 0) {
            return read;
        }

        allocHandle.lastBytesRead(nioData.position() - pos);
        buf.add(data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead()));
        free = false;
        return 1;
    } catch (Throwable cause) {
        PlatformDependent.throwException(cause);
        return -1;
    } finally {
        if (free) {
            data.release();
        }
    }
}
 

private V initialize(InternalThreadLocalMap threadLocalMap) {
    V v = null;
    try {
        v = initialValue();
    } catch (Exception e) {
        PlatformDependent.throwException(e);
    }

    threadLocalMap.setIndexedVariable(index, v);
    addToVariablesToRemove(threadLocalMap, this);
    return v;
}
 

@Override
public int forEachByteDesc(ByteProcessor processor) {
    ensureAccessible();
    try {
        return forEachByteDesc0(writerIndex - 1, readerIndex, processor);
    } catch (Exception e) {
        PlatformDependent.throwException(e);
        return -1;
    }
}
 

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    DatagramChannel ch = javaChannel();
    UkcpServerChannelConfig config = config();
    RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();

    ByteBuf data = allocHandle.allocate(config.getAllocator());
    allocHandle.attemptedBytesRead(data.writableBytes());
    boolean free = true;
    try {
        ByteBuffer nioData = data.internalNioBuffer(data.writerIndex(), data.writableBytes());
        int pos = nioData.position();
        InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(nioData);
        if (remoteAddress == null) {
            return 0;
        }

        allocHandle.lastBytesRead(nioData.position() - pos);
        buf.add(UkcpPacket.newInstance(data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead()),
                remoteAddress));
        free = false;
        return 1;
    } catch (Throwable cause) {
        PlatformDependent.throwException(cause);
        return -1;
    } finally {
        if (free) {
            data.release();
        }
    }
}
 

/**
 * Checks for the presence of an {@link Exception}.
 */
private ChannelFuture checkException(ChannelPromise promise) {
  Throwable t = lastException;
  if (t != null) {
    lastException = null;

    if (promise.isVoid()) {
        PlatformDependent.throwException(t);
    }

    return promise.setFailure(t);
  }

  return promise.setSuccess();
}
 

public void checkException() throws Exception {
    if (lastException == null) {
        return;
    }
    Throwable t = lastException;
    lastException = null;
    PlatformDependent.throwException(t);
}
 

@Override
protected void doBeginRead() throws Exception {
    if (readInProgress) {
        return;
    }

    ChannelPipeline pipeline = pipeline();
    Queue<Object> inboundBuffer = this.inboundBuffer;
    if (inboundBuffer.isEmpty()) {
        readInProgress = true;
        return;
    }

    final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
    final Integer stackDepth = threadLocals.localChannelReaderStackDepth();
    if (stackDepth < MAX_READER_STACK_DEPTH) {
        threadLocals.setLocalChannelReaderStackDepth(stackDepth + 1);
        try {
            for (;;) {
                Object received = inboundBuffer.poll();
                if (received == null) {
                    break;
                }
                pipeline.fireChannelRead(received);
            }
            pipeline.fireChannelReadComplete();
        } finally {
            threadLocals.setLocalChannelReaderStackDepth(stackDepth);
        }
    } else {
        try {
            eventLoop().execute(readTask);
        } catch (Throwable cause) {
            logger.warn("Closing Local channels {}-{} because exception occurred!", this, peer, cause);
            close();
            peer.close();
            PlatformDependent.throwException(cause);
        }
    }
}
 

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    DatagramChannel ch = javaChannel();
    DatagramChannelConfig config = config();
    RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();

    ByteBuf data = allocHandle.allocate(config.getAllocator());
    allocHandle.attemptedBytesRead(data.writableBytes());
    boolean free = true;
    try {
        ByteBuffer nioData = data.internalNioBuffer(data.writerIndex(), data.writableBytes());
        int pos = nioData.position();
        InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(nioData);
        if (remoteAddress == null) {
            return 0;
        }

        allocHandle.lastBytesRead(nioData.position() - pos);
        buf.add(new DatagramPacket(data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead()),
                localAddress(), remoteAddress));
        free = false;
        return 1;
    } catch (Throwable cause) {
        PlatformDependent.throwException(cause);
        return -1;
    }  finally {
        if (free) {
            data.release();
        }
    }
}
 

/**
 * Check if there was any {@link Throwable} received and if so rethrow it.
 */
public void checkException() {
    Throwable t = lastException;
    if (t == null) {
        return;
    }

    lastException = null;

    PlatformDependent.throwException(t);
}
 

@Override
protected int doReadMessages(List<Object> msgs) throws Exception {
    if (!readSelector.isOpen()) {
        return 0;
    }

    int readMessages = 0;

    final int selectedKeys = readSelector.select(SO_TIMEOUT);
    final boolean keysSelected = selectedKeys > 0;

    if (!keysSelected) {
        return readMessages;
    }
    // We must clear the selectedKeys because the Selector will never do it. If we do not clear it, the selectionKey
    // will always be returned even if there is no data can be read which causes performance issue. And in some
    // implementation of Selector, the select method may return 0 if the selectionKey which is ready for process has
    // already been in the selectedKeys and cause the keysSelected above to be false even if we actually have
    // something to read.
    readSelector.selectedKeys().clear();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    ByteBuf buffer = allocHandle.allocate(config().getAllocator());
    boolean free = true;

    try {
        ByteBuffer data = buffer.nioBuffer(buffer.writerIndex(), buffer.writableBytes());
        MessageInfo messageInfo = ch.receive(data, null, notificationHandler);
        if (messageInfo == null) {
            return readMessages;
        }

        data.flip();
        allocHandle.lastBytesRead(data.remaining());
        msgs.add(new SctpMessage(messageInfo,
                buffer.writerIndex(buffer.writerIndex() + allocHandle.lastBytesRead())));
        free = false;
        ++readMessages;
    } catch (Throwable cause) {
        PlatformDependent.throwException(cause);
    }  finally {
        if (free) {
            buffer.release();
        }
    }
    return readMessages;
}
 

@Override
    public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
        if (quietPeriod < 0) {
            throw new IllegalArgumentException("quietPeriod: " + quietPeriod + " (expected >= 0)");
        }
        if (timeout < quietPeriod) {
            throw new IllegalArgumentException(
                    "timeout: " + timeout + " (expected >= quietPeriod (" + quietPeriod + "))");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        if (isShuttingDown()) {
            return terminationFuture();
        }

        boolean inEventLoop = inEventLoop();
        boolean wakeup;
        int oldState;
        for (;;) {
            if (isShuttingDown()) {
                return terminationFuture();
            }
            int newState;
            wakeup = true;
            oldState = state;
//            如果是当前eventLoop直接修改状态为关闭
            if (inEventLoop) {
                newState = ST_SHUTTING_DOWN;
            } else {
                switch (oldState) {
//                    如果状态是没有启动或启动的直接修改为关闭
                    case ST_NOT_STARTED:
                    case ST_STARTED:
                        newState = ST_SHUTTING_DOWN;
                        break;
                    default:
                        newState = oldState;
                        wakeup = false;
                }
            }
//            修改线程执行器的状态
            if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
                break;
            }
        }
        gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
        gracefulShutdownTimeout = unit.toNanos(timeout);

//        如果线程执行器是没有启动
        if (oldState == ST_NOT_STARTED) {
            try {
//                如果是没有启动的开启线程执行事件监听
                doStartThread();
            } catch (Throwable cause) {
                STATE_UPDATER.set(this, ST_TERMINATED);
                terminationFuture.tryFailure(cause);

                if (!(cause instanceof Exception)) {
                    // Also rethrow as it may be an OOME for example
                    PlatformDependent.throwException(cause);
                }
                return terminationFuture;
            }
        }

//        如果需要唤醒线程组就唤醒
        if (wakeup) {
            wakeup(inEventLoop);
        }

        return terminationFuture();
    }
 

@Override
public ChannelFuture syncUninterruptibly() {
    PlatformDependent.throwException(cause);
    return this;
}