下面列出了io.netty.channel.Channel#writeAndFlush ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
/**
* 通过channel回执消息状态
*/
@Deprecated
protected void msgStatusResponse(Channel channel, Command command, String from, String to, String msgId,
long msgTime) {
CoreProto.MsgStatus status = CoreProto.MsgStatus.newBuilder().setMsgId(msgId).setMsgServerTime(msgTime)
.setMsgStatus(1).build();
ImStcMessageProto.MsgWithPointer statusMsg = ImStcMessageProto.MsgWithPointer.newBuilder()
.setType(MsgType.MSG_STATUS).setStatus(status).build();
ImStcMessageProto.ImStcMessageRequest request = ImStcMessageProto.ImStcMessageRequest.newBuilder()
.addList(statusMsg).build();
CoreProto.TransportPackageData data = CoreProto.TransportPackageData.newBuilder()
.setData(request.toByteString()).build();
channel.writeAndFlush(new RedisCommand().add(CommandConst.PROTOCOL_VERSION).add(CommandConst.IM_MSG_TOCLIENT)
.add(data.toByteArray()));
}
public void testLoginAndUserPasswordUpdate() throws Exception {
login();
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
while(ftdcChannel == null) {
ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
Thread.sleep(1000);
}
ByteBuf buffer = ftdcChannel.alloc().buffer();
ReqUserPasswordUpdate order = new ReqUserPasswordUpdate();
order.setUserID(userid);
order.setOldPassword("890619ab");
order.setNewPassword("890619aa");
order.setBrokerID(brokerid);
order.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), ri.getReqId(), TID.UserPasswordUpdateReq.id(), Sequence.UserPasswordUpdate);
ftdcChannel.writeAndFlush(ftdc);
Thread.sleep(WATI_TIME);
}
@Test(
alwaysRun = true,
dependsOnMethods = {"testPredictionsBinary"})
public void testPredictionsJson() throws InterruptedException {
Channel channel = TestUtils.getInferenceChannel(configManager);
TestUtils.setResult(null);
TestUtils.setLatch(new CountDownLatch(1));
DefaultFullHttpRequest req =
new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, HttpMethod.POST, "/predictions/noop");
req.content().writeCharSequence("{\"data\": \"test\"}", CharsetUtil.UTF_8);
HttpUtil.setContentLength(req, req.content().readableBytes());
req.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
channel.writeAndFlush(req);
TestUtils.getLatch().await();
Assert.assertEquals(TestUtils.getResult(), "OK");
}
public static void write(Channel channel, CommandResponse response) {
CoreProto.TransportPackageData.Builder packageDataBuilder = CoreProto.TransportPackageData.newBuilder();
CoreProto.ErrorInfo errorInfo = CoreProto.ErrorInfo.newBuilder().setCode(response.getErrCode())
.setInfo(String.valueOf(response.getErrInfo())).build();
packageDataBuilder.setErr(errorInfo);
Map<Integer, String> header = new HashMap<Integer, String>();
header.put(CoreProto.HeaderKey.SITE_SERVER_VERSION_VALUE, CommandConst.SITE_VERSION);
packageDataBuilder.putAllHeader(header);
if (response.getParams() != null) {
packageDataBuilder.setData(ByteString.copyFrom(response.getParams()));
}
channel.writeAndFlush(new RedisCommand().add(response.getVersion()).add(response.getAction())
.add(packageDataBuilder.build().toByteArray()));
}
@Override
public void reqUserLogout(RequestIdentity requestIdentity, FtdcReq userLogout) {
ApplicationRuntime.bindRequestIdentiity(requestIdentity);
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(requestIdentity.getBrokerId(), requestIdentity.getUserId());
if(ftdcChannel == null) {
getRequestSpi(requestIdentity);
return;
}
FtdcTraderSpi ftdcTraderSpi = getSpi(ftdcChannel);
if(ftdcChannel.isActive()) {
ByteBuf buffer = ftdcChannel.alloc().buffer();
userLogout.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), requestIdentity.getReqId(), TID.UserLogoutReq.id(), Sequence.UserLogout);
ftdcChannel.writeAndFlush(ftdc);
}else {
fireRspError(ftdcTraderSpi, requestIdentity);
}
}
@Override
public void handle(Channel channel, Command command, String... args) {
RegistryMonitor monitor = getParent().getRegistryMonitor();
if (monitor == null) {
return;
}
if (args.length < 4) {
channel.writeAndFlush("Args[2]: host, args[3]: port" + JConstants.NEWLINE);
return;
}
Command.ChildCommand childGrep = null;
if (args.length >= 6) {
childGrep = command.parseChild(args[4]);
}
for (String a : monitor.listServicesByAddress(args[2], Integer.parseInt(args[3]))) {
if (childGrep == Command.ChildCommand.GREP) {
if (a.contains(args[5])) {
channel.writeAndFlush(a + JConstants.NEWLINE);
}
} else {
channel.writeAndFlush(a + JConstants.NEWLINE);
}
}
}
public void testLoginAndQryOrder() throws Exception {
UserLoginReqhandler userLoginReqhandler = new UserLoginReqhandler();
CtpClient.ctp(host, port, userLoginReqhandler);
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
while(ftdcChannel == null) {
ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
Thread.sleep(1000);
}
ByteBuf buffer = ftdcChannel.alloc().buffer();
ReqQryOrder order = new ReqQryOrder();
order.setBrokerID(brokerid);
order.setInvestorID(userid);
order.setInsertTimeStart("20170928");
order.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), 6969, TID.QryOrderReq.id(), Sequence.QryOrder);
ftdcChannel.writeAndFlush(ftdc);
Thread.sleep(WATI_TIME);
}
public static void broadCastPing() {
try {
rwLock.readLock().lock();
logger.info("broadCastPing userCount: {}", userCount.intValue());
Set<Channel> keySet = userInfos.keySet();
for (Channel ch : keySet) {
UserInfo userInfo = userInfos.get(ch);
//如果channel是没有用户信息或者没有授权的用户则跳过
if (userInfo == null || !userInfo.isAuth()) {
continue;
}
ch.writeAndFlush(new TextWebSocketFrame(ChatProto.buildPingProto()));
}
} finally {
rwLock.readLock().unlock();
}
}
public void testQryTransferSerial() throws Exception {
login();
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
while(ftdcChannel == null) {
ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
Thread.sleep(1000);
}
ByteBuf buffer = ftdcChannel.alloc().buffer();
ReqQryTransferSerial reqQryTransferSerial = new ReqQryTransferSerial();
reqQryTransferSerial.setAccountID(userid);
// reqQryTransferSerial.setBankID("");
reqQryTransferSerial.setBrokerID(brokerid);
// reqQryTransferSerial.setCurrencyID("CNY");
reqQryTransferSerial.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), ri.getReqId(), TID.QryTransferSerialReq.id(), Sequence.QryTransferSerial);
ftdcChannel.writeAndFlush(ftdc);
Thread.sleep(WATI_TIME);
}
private static void writeAndVerifyReceived(ByteBuf message, Channel sendChannel, CountDownLatch receiverLatch,
MessageReceiver receiver) throws Exception {
List<ByteBuf> dataCapture = null;
try {
sendChannel.writeAndFlush(message);
receiverLatch.await(5, TimeUnit.SECONDS);
message.resetReaderIndex();
ArgumentCaptor<ByteBuf> captor = ArgumentCaptor.forClass(ByteBuf.class);
verify(receiver).messageReceived(captor.capture());
dataCapture = captor.getAllValues();
assertEquals(message, dataCapture.get(0));
} finally {
if (dataCapture != null) {
for (ByteBuf data : dataCapture) {
data.release();
}
}
}
}
@Test(
alwaysRun = true,
dependsOnMethods = {"testPredictionsInvalidRequestSize"})
public void testPredictionsValidRequestSize() throws InterruptedException {
Channel channel = TestUtils.getInferenceChannel(configManager);
TestUtils.setResult(null);
TestUtils.setLatch(new CountDownLatch(1));
DefaultFullHttpRequest req =
new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, HttpMethod.POST, "/predictions/noop");
req.content().writeZero(10385760);
HttpUtil.setContentLength(req, req.content().readableBytes());
req.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_OCTET_STREAM);
channel.writeAndFlush(req);
TestUtils.getLatch().await();
Assert.assertEquals(TestUtils.getHttpStatus(), HttpResponseStatus.OK);
}
@Override
public void handleRequest(Channel client, MqttMessage message) throws Exception {
String clientId = NettyAttrManager.getAttrClientId(client);
if (logger.isDebugEnabled()) {
logger.debug(String.format("PingRequest clientId:%s", clientId));
}
MqttFixedHeader pingHeader = new MqttFixedHeader(
MqttMessageType.PINGRESP,
false,
AT_MOST_ONCE,
false,
0);
MqttMessage pingResp = new MqttMessage(pingHeader);
client.writeAndFlush(pingResp);
}
public void testFromFutureToBank() throws Exception {
login();
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
while(ftdcChannel == null) {
ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
Thread.sleep(1000);
}
ByteBuf buffer = ftdcChannel.alloc().buffer();
ReqFromFutureToBank reqFromFutureToBank = new ReqFromFutureToBank();
reqFromFutureToBank.setBrokerID(brokerid);
reqFromFutureToBank.setRequestID(6972);
reqFromFutureToBank.setBankID("4");
reqFromFutureToBank.setBankBranchID("0000");
reqFromFutureToBank.setAccountID(userid);
reqFromFutureToBank.setPassword("241398");
reqFromFutureToBank.setTradeAmount(1);
reqFromFutureToBank.setBankPassWord("241398");
reqFromFutureToBank.setTradeCode(FtdcTradeCode.FutureToBank);
reqFromFutureToBank.setSecuPwdFlag(FtdcPwdFlag.FTDC_BlankCheck);
reqFromFutureToBank.setBankPwdFlag(FtdcPwdFlag.FTDC_NoCheck);
reqFromFutureToBank.setVerifyCertNoFlag(FtdcYesNoIndicator.TDC_No);
reqFromFutureToBank.setCurrencyID(FtdcCurrencyID.CNY);
reqFromFutureToBank.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), ri.getReqId(), TID.FromFutureToBankReq.id(), Sequence.FromFutureToBank);
ftdcChannel.writeAndFlush(ftdc);
Thread.sleep(WATI_TIME);
}
public void testLoginAndOrderInsert() throws Exception {
login();
Channel ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
while(ftdcChannel == null) {
ftdcChannel = ApplicationRuntime.getFtdcChannel(brokerid, userid);
Thread.sleep(1000);
}
ByteBuf buffer = ftdcChannel.alloc().buffer();
ReqInputOrder order = new ReqInputOrder();
order.setInstrumentID("ag1712");
order.setInvestorID(userid);
order.setDirection(FtdcDirection.BUY);
order.setOrderPriceType(FtdcOrderPriceType.FTDC_LimitPrice);
order.setCombOffsetFlag(FtdcOffsetFlagType.FTDC_Open);
order.setVolumeTotalOriginal(1);
order.setCombHedgeFlag(FtdcBillHedgeFlag.FTDC_Speculation);
order.setLimitPrice(3750);
order.setTimeCondition(FtdcTimeCondition.FTDC_GFD);
order.setVolumeCondition(FtdcVolumeCondition.HOST_AV);
order.setMinVolume(1);
order.setContingentCondition(FtdcContingentCondition.FTDC_Immediately);
order.setStopPrice(0);
order.setForceCloseReason(FtdcForceCLoseReson.THOST_FTDCFCC_NotForceClose);
order.setExchangeID(FtdcExchange.FTDC_SHFE);
order.setBrokerID(brokerid);
order.setRequestID(7000);
order.setOrderRef("101");
// order.setIPAddress("122.40.123.44");
order.write(buffer.retain());
FtdcProtocol ftdc = new FtdcProtocol(FtdType.FTDTypeCompressed, buffer, FtdcType.REQ.type(), ri.getReqId(), TID.OrderInsertReq.id(), Sequence.OrderInsert);
ftdcChannel.writeAndFlush(ftdc);
Thread.sleep(WATI_TIME);
}
private void testInvocationsJson(Channel channel) throws InterruptedException {
result = null;
latch = new CountDownLatch(1);
DefaultFullHttpRequest req =
new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, HttpMethod.POST, "/invocations?model_name=noop");
req.content().writeCharSequence("{\"data\": \"test\"}", CharsetUtil.UTF_8);
HttpUtil.setContentLength(req, req.content().readableBytes());
req.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
channel.writeAndFlush(req);
latch.await();
Assert.assertEquals(result, "OK");
}
private void testSyncScaleModel(Channel channel) throws InterruptedException {
result = null;
latch = new CountDownLatch(1);
HttpRequest req =
new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1,
HttpMethod.PUT,
"/models/noop_v0.1?synchronous=true&min_worker=1");
channel.writeAndFlush(req);
latch.await();
StatusResponse resp = JsonUtils.GSON.fromJson(result, StatusResponse.class);
Assert.assertEquals(resp.getStatus(), "Workers scaled");
}
private void handleTransferMessage(ChannelHandlerContext ctx, NatMessage msg) {
Channel natChannel = ctx.channel();
String clientId = natChannel.attr(Constant.NAT_CHANNEL_CLIENT_KEY).get();
if (clientId == null) {
//not happened
log.error("no client bound for channel:{}", natChannel);
ctx.close();
return;
}
long seq = msg.getSerialNumber();
NatClientImage client = clientManager.getClient(clientId);
if (client == null) {
log.error("now client registered for clientId:{}", clientId);
ctx.close();
return;
}
Channel userMappingChannel = client.queryUserMappingChannel(seq);
if (userMappingChannel == null) {
log.warn("can not find userMapping channel for request :{} client:{} ,send a close message to client endpoint", seq, client);
NatMessage natMessage = new NatMessage();
natMessage.setType(NatMessage.TYPE_DISCONNECT);
natMessage.setSerialNumber(seq);
natChannel.writeAndFlush(natMessage);
return;
}
log.info("forward data from nat client:{} to user endpoint with request:{}", clientId, seq);
byte[] data = msg.getData();
ByteBuf buf = ctx.alloc().buffer(data.length);
buf.writeBytes(data);
userMappingChannel.writeAndFlush(buf);
log.info("reply completed for clientId:{} for request:{}", clientId, seq);
}
public void processPingReq(Channel channel, MqttMessage msg){
MqttMessage pingRespMessage = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0),
null,
null);
log.info("PINGREQ - clientId: {}", (String) channel.attr(AttributeKey.valueOf("clientId")).get());
channel.writeAndFlush(pingRespMessage);
}
public static void p2p(Integer uid, String nick, String other, String message) {
if (!BlankUtil.isBlank(message)) {
try {
rwLock.readLock().lock();
message = "[来自于用户" + nick +"的消息]:" + message;
Set<Channel> keySet = userInfos.keySet();
for (Channel ch : keySet) {
UserInfo userInfo = userInfos.get(ch);
// 找出对应channel进行发送
if (userInfo == null || !userInfo.isAuth() || !userInfo.getNick().equals(other)) {
continue;
}
//在线用户的个人对个人通信直接走channel，不走第三方中间件和其它
ch.writeAndFlush(new TextWebSocketFrame(ChatProto.buildMessProto(userInfo.getId(), userInfo.getUsername(), message)));
return;
}
LambdaQueryWrapper<Account> lambdaQueryWrapper = new LambdaQueryWrapper<>();
lambdaQueryWrapper.eq(Account::getUsername, other);
Account account = accountMapperStatic.selectOne(lambdaQueryWrapper);
if (account != null) {
offlineInfoTransmitStatic.pushP2P(account.getId(), message);
}
} finally {
rwLock.readLock().unlock();
}
}
}
/**
* AuthenticationCleartextPassword (B)
*
* Byte1('R')
* Identifies the message as an authentication request.
*
* Int32(8)
* Length of message contents in bytes, including self.
*
* Int32(3)
* Specifies that a clear-text password is required.
*
* @param channel The channel to write to.
*/
static void sendAuthenticationCleartextPassword(Channel channel) {
ByteBuf buffer = channel.alloc().buffer(9);
buffer.writeByte('R');
buffer.writeInt(8);
buffer.writeInt(3);
ChannelFuture channelFuture = channel.writeAndFlush(buffer);
if (LOGGER.isTraceEnabled()) {
channelFuture.addListener((ChannelFutureListener) future -> LOGGER.trace("sentAuthenticationCleartextPassword"));
}
}