io.netty.channel.Channel#write ( )源码实例Demo

@Ignore("not supported anymore atm")
@Test
public void cancellingWritesBeforeFlush() {
    LastInboundHandler inboundHandler = streamActiveAndWriteHeaders(inboundStream);
    Channel childChannel = inboundHandler.channel();

    Http2HeadersFrame headers1 = new DefaultHttp2HeadersFrame(new DefaultHttp2Headers());
    Http2HeadersFrame headers2 = new DefaultHttp2HeadersFrame(new DefaultHttp2Headers());
    ChannelPromise writePromise = childChannel.newPromise();
    childChannel.write(headers1, writePromise);
    childChannel.write(headers2);
    assertTrue(writePromise.cancel(false));
    childChannel.flush();

    Http2HeadersFrame headers = parentChannel.readOutbound();
    assertSame(headers, headers2);
}
 

@Override
public void writeReadingCommand(Channel channel) {
    if (isNotValid()) {
        return;
    }
    StringMessage msg = makeUTF8StringMessage(HARDWARE, READING_MSG_ID,
            DataStream.makeReadingHardwareBody(pinType.pintTypeChar, pin));
    channel.write(msg, channel.voidPromise());
}
 

/**
 * Return that we need cannot not support the web socket version
 */
public static ChannelFuture sendUnsupportedVersionResponse(Channel channel, ChannelPromise promise) {
    HttpResponse res = new DefaultHttpResponse(
            HttpVersion.HTTP_1_1,
            HttpResponseStatus.UPGRADE_REQUIRED);
    res.headers().set(Names.SEC_WEBSOCKET_VERSION, WebSocketVersion.V13.toHttpHeaderValue());
    return channel.write(res, promise);
}
 

@Override
public void sendAppSync(Channel appChannel, int dashId, int targetId, boolean useNewSyncFormat) {
    if (isNotValid() || lastCommands.size() == 0) {
        return;
    }
    if (targetId == ANY_TARGET || this.deviceId == targetId) {
        for (String storedValue : lastCommands) {
            String body = prependDashIdAndDeviceId(dashId, deviceId, makeHardwareBody(pinType, pin, storedValue));
            appChannel.write(makeUTF8StringMessage(APP_SYNC, SYNC_DEFAULT_MESSAGE_ID, body));
        }
    }
}
 

private void completeLogin(Channel channel, Session session, User user,
                                  DashBoard dash, Device device, int msgId) {
    log.debug("completeLogin. {}", channel);

    session.addHardChannel(channel);
    channel.write(ok(msgId));

    String body = dash.buildPMMessage(device.id);
    if (dash.isActive && body != null) {
        channel.write(makeASCIIStringMessage(HARDWARE, HARDWARE_PIN_MODE_MSG_ID, body));
    }

    channel.flush();

    String responseBody = String.valueOf(dash.id) + DEVICE_SEPARATOR + device.id;
    session.sendToApps(HARDWARE_CONNECTED, msgId, dash.id, responseBody);
    log.trace("Connected device id {}, dash id {}", device.id, dash.id);
    device.connected();
    if (device.firstConnectTime == 0) {
        device.firstConnectTime = device.connectTime;
    }
    if (allowStoreIp) {
        device.lastLoggedIP = IPUtils.getIp(channel.remoteAddress());
    }

    log.info("{} hardware joined.", user.email);
}
 

@Override
public void writeReadingCommand(Channel channel) {
    if (dataStreams == null) {
        return;
    }
    for (DataStream dataStream : dataStreams) {
        if (dataStream.isValid()) {
            StringMessage msg = makeUTF8StringMessage(HARDWARE, READING_MSG_ID,
                    DataStream.makeReadingHardwareBody(dataStream.pinType.pintTypeChar, dataStream.pin));
            channel.write(msg, channel.voidPromise());
        }
    }
}
 

@Override
protected void onMessageReceived(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
    // directly write getFrontendChannel data to getMysqlChannel real mysql connection
    ByteBuf userByteBuf = ctx.alloc().heapBuffer(msg.readableBytes());
    msg.getBytes(0,userByteBuf);

    Channel channel = channelSupplier.get();
    setReadyWritePacket(channel,msg);
    ctx.fireChannelRead(userByteBuf);
    ByteBuf readyWritePacket = getReadyWritePacket(channel);
    if(readyWritePacket != null) {
        channel.write(readyWritePacket);
    }
}
 

/**
	 * Write to channel.
	 *
	 * @param channel the channel
	 * @param magicBytes the magic bytes
	 * @param pathOrCommand the path or command
	 * @param attachment the attachment
	 */
	private void writeToChannel(Channel channel, String[] magicBytes, Object pathOrCommand, Object attachment) throws ConnectException {

		//update leader agent details
//		if (pathOrCommand instanceof CommandWritable
//				&& ((CommandWritable) pathOrCommand).isCommandForMaster()) {
//			CommandWritable commandWritable = (CommandWritable) pathOrCommand;
//			AgentNode agent = ZKUtils.getLeaderAgentfromZK();
//			commandWritable.setNameNodeHost(agent.getHost());
//			if(commandWritable.isHasSshAuthKeysFile()){
//				commandWritable.setSshAuthKeysFile(agent.getPrivateKey());
//			}
//		}

		long firstAttempt = System.currentTimeMillis();
		long timeOut = RemotingConstants.TEN * RemotingConstants.THOUSAND;
		while (!channel.isOpen() || !channel.isActive()) {
			if (System.currentTimeMillis() - firstAttempt >= timeOut) {
				try {
					throw new TimeoutException();
				} catch (TimeoutException e) {
					logger.error("Waited for 10 sec for connection reattempt to JumbuneAgent, but failed to connect", e);
				}
				break;
			}
		}
		if (!channel.isActive()) {
			logger.warn("channel #" + channel.hashCode() + " still disconnected, about to write on disconnected Channel");
		}
		if (attachment != null && attachment instanceof CyclicBarrier) {
			channel.attr(RemotingConstants.barrierKey).set((CyclicBarrier)attachment);
		}else if (attachment != null) {
			channel.attr(RemotingConstants.handlerKey).set((ChannelInboundHandler)attachment);
		}
		channel.write(Unpooled.wrappedBuffer(magicBytes[0].getBytes(), magicBytes[1].getBytes(), magicBytes[2].getBytes()));
		channel.write(pathOrCommand);
		channel.flush();
	}
 

@Override
public final void run(Channel channel) {
  channel.write(this, promise);
}
 

public void outboundReceived(Channel channel, Packet packet) {
	if(this.outboundChannel == channel && this.redirecting) {
		return;
	}
	switch(packet.getOpcode()) {
	case PlayJoinGamePacket.opcode:
		PlayJoinGamePacket playJoinGamePacket = (PlayJoinGamePacket) packet;
		if(this.config.proxy_isPlayerTab()) {
			playJoinGamePacket.setMaxPlayers(60);
		} else {
			playJoinGamePacket.setMaxPlayers(0);
		}
		if(this.clientSettingsPacket != null) {
			channel.write(this.clientSettingsPacket);
		}
		this.serverEntityId = playJoinGamePacket.getEntityId();
		if(this.state == ProxyState.INIT) {
			this.clientEntityId = playJoinGamePacket.getEntityId();
		} else {
			this.inboundChannel.write(new PlayRespawnPacket(playJoinGamePacket.getDimension() == 0 ? 1 : 0, 2, 0, "DEFAULT"));
			this.inboundChannel.write(new PlayRespawnPacket(playJoinGamePacket.getDimension(), playJoinGamePacket.getDifficulty(), playJoinGamePacket.getGamemode(), playJoinGamePacket.getLevelType()));
			Iterator<String> playersTabbed = this.playersTabbed.iterator();
			while(playersTabbed.hasNext()) {
				this.inboundChannel.write(new PlayPlayerListPacket(playersTabbed.next(), false, 0));
				playersTabbed.remove();
			}
			Iterator<String> scoreboards = this.scoreboards.iterator();
			while(scoreboards.hasNext()) {
				this.inboundChannel.write(new PlayScoreObjectivePacket(scoreboards.next(), "", 1));
				scoreboards.remove();
			}
			Iterator<String> teams = this.teams.iterator();
			while(teams.hasNext()) {
				this.inboundChannel.write(new PlayTeamPacket(teams.next(), 1, null, null, null, 0, null));
				teams.remove();
			}
			this.inboundChannel.flush();
			return;
		}
		break;
	case PlayPlayerListPacket.opcode:
		PlayPlayerListPacket playPlayerListPacket = (PlayPlayerListPacket) packet;
		if(playPlayerListPacket.isOnline()) {
			this.playersTabbed.add(playPlayerListPacket.getName());
		} else {
			this.playersTabbed.remove(playPlayerListPacket.getName());
		}
		break;
	case PlayScoreObjectivePacket.opcode:
		PlayScoreObjectivePacket playScoreObjectivePacket = (PlayScoreObjectivePacket) packet;
		if(playScoreObjectivePacket.isCreating()) {
			this.scoreboards.add(playScoreObjectivePacket.getName());
		} else if(playScoreObjectivePacket.isRemoving()) {
			this.scoreboards.remove(playScoreObjectivePacket.getName());
		}
		break;
	case PlayTeamPacket.opcode:
		PlayTeamPacket teamPacket = (PlayTeamPacket) packet;
		if(teamPacket.isCreating()) {
			this.teams.add(teamPacket.getName());
		} else if(teamPacket.isRemoving()) {
			this.teams.remove(teamPacket.getName());
		}
		break;
	case PlayDisconnectPacket.opcode:
		this.disconnect(packet);
		return;
	default:
		if(packet instanceof GenericPacket) {
			((GenericPacket) packet).swapEntityId(this.clientEntityId, this.serverEntityId);
		}
		break;
	}
	this.inboundChannel.writeAndFlush(packet);
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object message) throws Exception {
    LOG.trace("New data read: incoming: {}", message);

    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch, (FullHttpResponse) message);
        LOG.trace("WebSocket Client connected! {}", ctx.channel());
        // Now trigger super processing as we are really connected.
        if(handshakeTimeoutFuture.cancel(false)) {
            NettyWsTransport.super.handleConnected(ch);
        }
        return;
    }

    // We shouldn't get this since we handle the handshake previously.
    if (message instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) message;
        throw new IllegalStateException(
            "Unexpected FullHttpResponse (getStatus=" + response.status() +
            ", content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) message;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        LOG.warn("WebSocket Client received message: " + textFrame.text());
        ctx.fireExceptionCaught(new IOException("Received invalid frame over WebSocket."));
    } else if (frame instanceof BinaryWebSocketFrame) {
        BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
        LOG.trace("WebSocket Client received data: {} bytes", binaryFrame.content().readableBytes());
        listener.onData(binaryFrame.content());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
        LOG.trace("WebSocket Client received data continuation: {} bytes", continuationFrame.content().readableBytes());
        listener.onData(continuationFrame.content());
    } else if (frame instanceof PingWebSocketFrame) {
        LOG.trace("WebSocket Client received ping, response with pong");
        ch.write(new PongWebSocketFrame(frame.content()));
    } else if (frame instanceof CloseWebSocketFrame) {
        LOG.trace("WebSocket Client received closing");
        ch.close();
    }
}
 

@Override
public void send(Channel channel, TftpPacket packet) throws Exception {
    packet.setRemoteAddress(getRemoteAddress());
    channel.write(packet, channel.voidPromise());
}
 

/**
 * Multipart example
 */
private static void formpostmultipart(
        Bootstrap bootstrap, String host, int port, URI uriFile, HttpDataFactory factory,
        List<Entry<String, String>> headers, List<InterfaceHttpData> bodylist) throws Exception {
    // XXX /formpostmultipart
    // Start the connection attempt.
    ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
    // Wait until the connection attempt succeeds or fails.
    Channel channel = future.sync().channel();

    // Prepare the HTTP request.
    HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uriFile.toASCIIString());

    // Use the PostBody encoder
    HttpPostRequestEncoder bodyRequestEncoder =
            new HttpPostRequestEncoder(factory, request, true); // true => multipart

    // it is legal to add directly header or cookie into the request until finalize
    for (Entry<String, String> entry : headers) {
        request.headers().set(entry.getKey(), entry.getValue());
    }

    // add Form attribute from previous request in formpost()
    bodyRequestEncoder.setBodyHttpDatas(bodylist);

    // finalize request
    bodyRequestEncoder.finalizeRequest();

    // send request
    channel.write(request);

    // test if request was chunked and if so, finish the write
    if (bodyRequestEncoder.isChunked()) {
        channel.write(bodyRequestEncoder);
    }
    channel.flush();

    // Now no more use of file representation (and list of HttpData)
    bodyRequestEncoder.cleanFiles();

    // Wait for the server to close the connection.
    channel.closeFuture().sync();
}
 

/**
 * | 'S' | int32 len | str name | str value
 * <p>
 * See https://www.postgresql.org/docs/9.2/static/protocol-flow.html#PROTOCOL-ASYNC
 * <p>
 * > At present there is a hard-wired set of parameters for which ParameterStatus will be generated: they are
 * <p>
 * - server_version,
 * - server_encoding,
 * - client_encoding,
 * - application_name,
 * - is_superuser,
 * - session_authorization,
 * - DateStyle,
 * - IntervalStyle,
 * - TimeZone,
 * - integer_datetimes,
 * - standard_conforming_string
 */
static void sendParameterStatus(Channel channel, final String name, final String value) {
    byte[] nameBytes = name.getBytes(StandardCharsets.UTF_8);
    byte[] valueBytes = value.getBytes(StandardCharsets.UTF_8);

    int length = 4 + nameBytes.length + 1 + valueBytes.length + 1;
    ByteBuf buffer = channel.alloc().buffer(length + 1);
    buffer.writeByte('S');
    buffer.writeInt(length);
    writeCString(buffer, nameBytes);
    writeCString(buffer, valueBytes);
    ChannelFuture channelFuture = channel.write(buffer);
    if (LOGGER.isTraceEnabled()) {
        channelFuture.addListener((ChannelFutureListener) future -> LOGGER.trace("sentParameterStatus {}={}", name, value));
    }
}
 

/**
 * Standard post without multipart but already support on Factory (memory management)
 *
 * @return the list of HttpData object (attribute and file) to be reused on next post
 */
private static List<InterfaceHttpData> formpost(
        Bootstrap bootstrap,
        String host, int port, URI uriSimple, File file, HttpDataFactory factory,
        List<Entry<String, String>> headers) throws Exception {
    // XXX /formpost
    // Start the connection attempt.
    ChannelFuture future = bootstrap.connect(SocketUtils.socketAddress(host, port));
    // Wait until the connection attempt succeeds or fails.
    Channel channel = future.sync().channel();

    // Prepare the HTTP request.
    HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uriSimple.toASCIIString());

    // Use the PostBody encoder
    HttpPostRequestEncoder bodyRequestEncoder =
            new HttpPostRequestEncoder(factory, request, false);  // false => not multipart

    // it is legal to add directly header or cookie into the request until finalize
    for (Entry<String, String> entry : headers) {
        request.headers().set(entry.getKey(), entry.getValue());
    }

    // add Form attribute
    bodyRequestEncoder.addBodyAttribute("getform", "POST");
    bodyRequestEncoder.addBodyAttribute("info", "first value");
    bodyRequestEncoder.addBodyAttribute("secondinfo", "secondvalue ���&");
    bodyRequestEncoder.addBodyAttribute("thirdinfo", textArea);
    bodyRequestEncoder.addBodyAttribute("fourthinfo", textAreaLong);
    bodyRequestEncoder.addBodyFileUpload("myfile", file, "application/x-zip-compressed", false);

    // finalize request
    request = bodyRequestEncoder.finalizeRequest();

    // Create the bodylist to be reused on the last version with Multipart support
    List<InterfaceHttpData> bodylist = bodyRequestEncoder.getBodyListAttributes();

    // send request
    channel.write(request);

    // test if request was chunked and if so, finish the write
    if (bodyRequestEncoder.isChunked()) { // could do either request.isChunked()
        // either do it through ChunkedWriteHandler
        channel.write(bodyRequestEncoder);
    }
    channel.flush();

    // Do not clear here since we will reuse the InterfaceHttpData on the next request
    // for the example (limit action on client side). Take this as a broadcast of the same
    // request on both Post actions.
    //
    // On standard program, it is clearly recommended to clean all files after each request
    // bodyRequestEncoder.cleanFiles();

    // Wait for the server to close the connection.
    channel.closeFuture().sync();
    return bodylist;
}
 

/**
 * Multipart example
 */
private static void formpostmultipart(
        Bootstrap bootstrap, String host, int port, URI uriFile, HttpDataFactory factory,
        Iterable<Entry<String, String>> headers, List<InterfaceHttpData> bodylist) throws Exception {
    // XXX /formpostmultipart
    // Start the connection attempt.
    ChannelFuture future = bootstrap.connect(SocketUtils.socketAddress(host, port));
    // Wait until the connection attempt succeeds or fails.
    Channel channel = future.sync().channel();

    // Prepare the HTTP request.
    HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uriFile.toASCIIString());

    // Use the PostBody encoder
    HttpPostRequestEncoder bodyRequestEncoder =
            new HttpPostRequestEncoder(factory, request, true); // true => multipart

    // it is legal to add directly header or cookie into the request until finalize
    for (Entry<String, String> entry : headers) {
        request.headers().set(entry.getKey(), entry.getValue());
    }

    // add Form attribute from previous request in formpost()
    bodyRequestEncoder.setBodyHttpDatas(bodylist);

    // finalize request
    bodyRequestEncoder.finalizeRequest();

    // send request
    channel.write(request);

    // test if request was chunked and if so, finish the write
    if (bodyRequestEncoder.isChunked()) {
        channel.write(bodyRequestEncoder);
    }
    channel.flush();

    // Now no more use of file representation (and list of HttpData)
    bodyRequestEncoder.cleanFiles();

    // Wait for the server to close the connection.
    channel.closeFuture().sync();
}
 

public static void write(Channel channel, List<AmqpDeliverMessage> pendingMessages) {
    for (AmqpDeliverMessage message : pendingMessages) {
        channel.write(message);
    }
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object message) throws Exception {
   LOG.trace("New data read: incoming: {}", message);

   Channel ch = ctx.channel();
   if (!handshaker.isHandshakeComplete()) {
      handshaker.finishHandshake(ch, (FullHttpResponse) message);
      LOG.trace("WebSocket Client connected! {}", ctx.channel());
      // Now trigger super processing as we are really connected.
      NettyWSTransport.super.handleConnected(ch);
      return;
   }

   // We shouldn't get this since we handle the handshake previously.
   if (message instanceof FullHttpResponse) {
      FullHttpResponse response = (FullHttpResponse) message;
      throw new IllegalStateException(
         "Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
   }

   WebSocketFrame frame = (WebSocketFrame) message;
   if (frame instanceof TextWebSocketFrame) {
      TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
      LOG.warn("WebSocket Client received message: " + textFrame.text());
      ctx.fireExceptionCaught(new IOException("Received invalid frame over WebSocket."));
   } else if (frame instanceof BinaryWebSocketFrame) {
      BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
      LOG.trace("WebSocket Client received data: {} bytes", binaryFrame.content().readableBytes());
      listener.onData(binaryFrame.content());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
      LOG.trace("WebSocket Client received data continuation: {} bytes", continuationFrame.content().readableBytes());
      listener.onData(continuationFrame.content());
   } else if (frame instanceof PingWebSocketFrame) {
      LOG.trace("WebSocket Client received ping, response with pong");
      ch.write(new PongWebSocketFrame(frame.content()));
   } else if (frame instanceof CloseWebSocketFrame) {
      LOG.trace("WebSocket Client received closing");
      ch.close();
   }
}
 

public void testCancelWrite(ServerBootstrap sb, Bootstrap cb) throws Throwable {
    final TestHandler sh = new TestHandler();
    final TestHandler ch = new TestHandler();
    final ByteBuf a = Unpooled.buffer().writeByte('a');
    final ByteBuf b = Unpooled.buffer().writeByte('b');
    final ByteBuf c = Unpooled.buffer().writeByte('c');
    final ByteBuf d = Unpooled.buffer().writeByte('d');
    final ByteBuf e = Unpooled.buffer().writeByte('e');

    cb.handler(ch);
    sb.childHandler(sh);

    Channel sc = sb.bind().sync().channel();
    Channel cc = cb.connect().sync().channel();

    ChannelFuture f = cc.write(a);
    assertTrue(f.cancel(false));
    cc.writeAndFlush(b);
    cc.write(c);
    ChannelFuture f2 = cc.write(d);
    assertTrue(f2.cancel(false));
    cc.writeAndFlush(e);

    while (sh.counter < 3) {
        if (sh.exception.get() != null) {
            break;
        }
        if (ch.exception.get() != null) {
            break;
        }
        try {
            Thread.sleep(50);
        } catch (InterruptedException ignore) {
            // Ignore.
        }
    }
    sh.channel.close().sync();
    ch.channel.close().sync();
    sc.close().sync();

    if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
        throw sh.exception.get();
    }
    if (sh.exception.get() != null) {
        throw sh.exception.get();
    }
    if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
        throw ch.exception.get();
    }
    if (ch.exception.get() != null) {
        throw ch.exception.get();
    }
    assertEquals(0, ch.counter);
    assertEquals(Unpooled.wrappedBuffer(new byte[]{'b', 'c', 'e'}), sh.received);
}
 

/**
 * ParameterDescription (B)
 *
 *     Byte1('t')
 *
 *         Identifies the message as a parameter description.
 *     Int32
 *
 *         Length of message contents in bytes, including self.
 *     Int16
 *
 *         The number of parameters used by the statement (can be zero).
 *
 *     Then, for each parameter, there is the following:
 *
 *     Int32
 *
 *         Specifies the object ID of the parameter data type.
 *
 * @param channel The channel to write the parameter description to.
 * @param parameters A {@link SortedSet} containing the parameters from index 1 upwards.
 */
static void sendParameterDescription(Channel channel, DataType[] parameters) {
    final int messageByteSize = 4 + 2 + parameters.length * 4;
    ByteBuf buffer = channel.alloc().buffer(messageByteSize);
    buffer.writeByte('t');
    buffer.writeInt(messageByteSize);
    if (parameters.length > Short.MAX_VALUE) {
        buffer.release();
        throw new IllegalArgumentException("Too many parameters. Max supported: " + Short.MAX_VALUE);
    }
    buffer.writeShort(parameters.length);
    for (DataType dataType : parameters) {
        int pgTypeId = PGTypes.get(dataType).oid();
        buffer.writeInt(pgTypeId);
    }
    channel.write(buffer);
}