@Override
    public void send(Object message, boolean sent) throws RemotingException {
        // whether the channel is closed
        super.send(message, sent);

        boolean success = true;
        int timeout = 0;
        try {
            Object outputMessage = message;
            if (!encodeInIOThread) {
                ByteBuf buf = channel.alloc().buffer();
                ChannelBuffer buffer = new NettyBackedChannelBuffer(buf);
                codec.encode(this, buffer, message);
                outputMessage = buf;
            }
            ChannelFuture future = writeQueue.enqueue(outputMessage).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!(message instanceof Request)) {
                        return;
                    }
                    ChannelHandler handler = getChannelHandler();
                    if (future.isSuccess()) {
                        handler.sent(NettyChannel.this, message);
                    } else {
                        Throwable t = future.cause();
                        if (t == null) {
                            return;
                        }
                        Response response = buildErrorResponse((Request) message, t);
                        handler.received(NettyChannel.this, response);
                    }
                }
            });

            if (sent) {
                // wait timeout ms
                timeout = getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
                success = future.await(timeout);
            }
            Throwable cause = future.cause();
            if (cause != null) {
                throw cause;
            }
        } catch (Throwable e) {
            removeChannelIfDisconnected(channel);
            throw new RemotingException(
                    this,
                    "Failed to send message " + PayloadDropper.getRequestWithoutData(message) + " to "
                            + getRemoteAddress() + ", cause: " + e.getMessage(),
                    e);
        }
        if (!success) {
            throw new RemotingException(
                    this,
                    "Failed to send message " + PayloadDropper.getRequestWithoutData(message) + " to "
                            + getRemoteAddress() + "in timeout(" + timeout + "ms) limit");
        }
    }
@Test
    void testSend() throws Exception {
        Mockito.when(channel.eventLoop()).thenReturn(Mockito.mock(EventLoop.class));
        Mockito.when(channel.alloc()).thenReturn(PooledByteBufAllocator.DEFAULT);
        NettyChannel nettyChannel = NettyChannel.getOrAddChannel(channel, url, channelHandler);
        ChannelPromise future = Mockito.mock(ChannelPromise.class);
        Mockito.when(future.await(1000)).thenReturn(true);
        Mockito.when(future.cause()).thenReturn(null);
        Mockito.when(channel.writeAndFlush(Mockito.any())).thenReturn(future);
        Mockito.when(channel.newPromise()).thenReturn(future);
        Mockito.when(future.addListener(Mockito.any())).thenReturn(future);
        nettyChannel.send("msg", true);

        NettyChannel finalNettyChannel = nettyChannel;
        Exception exception = Mockito.mock(Exception.class);
        Mockito.when(exception.getMessage()).thenReturn("future cause");
        Mockito.when(future.cause()).thenReturn(exception);
        Assertions.assertThrows(
                RemotingException.class,
                () -> {
                    finalNettyChannel.send("msg", true);
                },
                "future cause");

        Mockito.when(future.await(1000)).thenReturn(false);
        Mockito.when(future.cause()).thenReturn(null);
        Assertions.assertThrows(
                RemotingException.class,
                () -> {
                    finalNettyChannel.send("msg", true);
                },
                "in timeout(1000ms) limit");

        ChannelPromise channelPromise = Mockito.mock(ChannelPromise.class);
        Mockito.when(channel.newPromise()).thenReturn(channelPromise);
        Mockito.when(channelPromise.await(1000)).thenReturn(true);
        Mockito.when(channelPromise.cause()).thenReturn(null);
        Mockito.when(channelPromise.addListener(Mockito.any())).thenReturn(channelPromise);
        finalNettyChannel.send("msg", true);
        ArgumentCaptor<GenericFutureListener> listenerArgumentCaptor =
                ArgumentCaptor.forClass(GenericFutureListener.class);
        Mockito.verify(channelPromise, Mockito.times(1)).addListener(listenerArgumentCaptor.capture());
    }