public class SessionHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class);
private CanalServerWithEmbedded embeddedServer;
public SessionHandler(){
}
public SessionHandler(CanalServerWithEmbedded embeddedServer){
this.embeddedServer = embeddedServer;
}
@SuppressWarnings({ "deprecation" })
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
logger.info("message receives in session handler...");
long start = System.nanoTime();
ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
ClientIdentity clientIdentity = null;
try {
switch (packet.getType()) {
case SUBSCRIPTION:
Sub sub = Sub.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(sub.getDestination()) && StringUtils.isNotEmpty(sub.getClientId())) {
clientIdentity = new ClientIdentity(sub.getDestination(),
Short.valueOf(sub.getClientId()),
sub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) {
runningMonitor.start();
}
}
embeddedServer.subscribe(clientIdentity);
byte[] ackBytes = NettyUtils.ackPacket();
NettyUtils.write(ctx.getChannel(), ackBytes, new ChannelFutureAggregator(sub.getDestination(),
sub,
packet.getType(),
ackBytes.length,
System.nanoTime() - start));
} else {
byte[] errorBytes = NettyUtils.errorPacket(401,
MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(sub.getDestination(),
sub,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 401));
}
break;
case UNSUBSCRIPTION:
Unsub unsub = Unsub.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(unsub.getDestination()) && StringUtils.isNotEmpty(unsub.getClientId())) {
clientIdentity = new ClientIdentity(unsub.getDestination(),
Short.valueOf(unsub.getClientId()),
unsub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
embeddedServer.unsubscribe(clientIdentity);
stopCanalInstanceIfNecessary(clientIdentity);
byte[] ackBytes = NettyUtils.ackPacket();
NettyUtils.write(ctx.getChannel(),
ackBytes,
new ChannelFutureAggregator(unsub.getDestination(),
unsub,
packet.getType(),
ackBytes.length,
System.nanoTime() - start));
} else {
byte[] errorBytes = NettyUtils.errorPacket(401,
MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(unsub.getDestination(),
unsub,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 401));
}
break;
case GET:
Get get = CanalPacket.Get.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(get.getDestination()) && StringUtils.isNotEmpty(get.getClientId())) {
clientIdentity = new ClientIdentity(get.getDestination(), Short.valueOf(get.getClientId()));
MDC.put("destination", clientIdentity.getDestination());
Message message = null;
if (get.getTimeout() == -1) {
message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
} else {
TimeUnit unit = convertTimeUnit(get.getUnit());
message = embeddedServer.getWithoutAck(clientIdentity,
get.getFetchSize(),
get.getTimeout(),
unit);
}
if (message.getId() != -1 && message.isRaw()) {
List<ByteString> rowEntries = message.getRawEntries();
int messageSize = 0;
messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId());
int dataSize = 0;
for (ByteString rowEntry : rowEntries) {
dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry);
}
messageSize += dataSize;
messageSize += 1 * rowEntries.size();
int size = 0;
size += com.google.protobuf.CodedOutputStream.computeEnumSize(3,
PacketType.MESSAGES.getNumber());
size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
+ com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
+ messageSize;
byte[] body = new byte[size];
CodedOutputStream output = CodedOutputStream.newInstance(body);
output.writeEnum(3, PacketType.MESSAGES.getNumber());
output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
output.writeRawVarint32(messageSize);
output.writeInt64(1, message.getId());
for (ByteString rowEntry : rowEntries) {
output.writeBytes(2, rowEntry);
}
output.checkNoSpaceLeft();
NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
get,
packet.getType(),
body.length,
System.nanoTime() - start,
message.getId() == -1));
} else {
Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
packetBuilder.setType(PacketType.MESSAGES).setVersion(NettyUtils.VERSION);
Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
messageBuilder.setBatchId(message.getId());
if (message.getId() != -1) {
if (message.isRaw() && !CollectionUtils.isEmpty(message.getRawEntries())) {
messageBuilder.addAllMessages(message.getRawEntries());
} else if (!CollectionUtils.isEmpty(message.getEntries())) {
for (Entry entry : message.getEntries()) {
messageBuilder.addMessages(entry.toByteString());
}
}
}
byte[] body = packetBuilder.setBody(messageBuilder.build().toByteString())
.build()
.toByteArray();
NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
get,
packet.getType(),
body.length,
System.nanoTime() - start,
message.getId() == -1));
}
} else {
byte[] errorBytes = NettyUtils.errorPacket(401,
MessageFormatter.format("destination or clientId is null", get.toString()).getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(get.getDestination(),
get,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 401));
}
break;
case CLIENTACK:
ClientAck ack = CanalPacket.ClientAck.parseFrom(packet.getBody());
MDC.put("destination", ack.getDestination());
if (StringUtils.isNotEmpty(ack.getDestination()) && StringUtils.isNotEmpty(ack.getClientId())) {
if (ack.getBatchId() == 0L) {
byte[] errorBytes = NettyUtils.errorPacket(402,
MessageFormatter.format("batchId should assign value", ack.toString()).getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(ack.getDestination(),
ack,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 402));
} else if (ack.getBatchId() == -1L) {
} else {
clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
embeddedServer.ack(clientIdentity, ack.getBatchId());
new ChannelFutureAggregator(ack.getDestination(),
ack,
packet.getType(),
0,
System.nanoTime() - start).operationComplete(null);
}
} else {
byte[] errorBytes = NettyUtils.errorPacket(401,
MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(ack.getDestination(),
ack,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 401));
}
break;
case CLIENTROLLBACK:
ClientRollback rollback = CanalPacket.ClientRollback.parseFrom(packet.getBody());
MDC.put("destination", rollback.getDestination());
if (StringUtils.isNotEmpty(rollback.getDestination())
&& StringUtils.isNotEmpty(rollback.getClientId())) {
clientIdentity = new ClientIdentity(rollback.getDestination(),
Short.valueOf(rollback.getClientId()));
if (rollback.getBatchId() == 0L) {
embeddedServer.rollback(clientIdentity);
} else {
embeddedServer.rollback(clientIdentity, rollback.getBatchId());
}
new ChannelFutureAggregator(rollback.getDestination(),
rollback,
packet.getType(),
0,
System.nanoTime() - start).operationComplete(null);
} else {
byte[] errorBytes = NettyUtils.errorPacket(401,
MessageFormatter.format("destination or clientId is null", rollback.toString())
.getMessage());
NettyUtils.write(ctx.getChannel(),
errorBytes,
new ChannelFutureAggregator(rollback.getDestination(),
rollback,
packet.getType(),
errorBytes.length,
System.nanoTime() - start,
(short) 401));
}
break;
default:
byte[] errorBytes = NettyUtils.errorPacket(400,
MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage());
NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
.getRemoteAddress()
.toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
break;
}
} catch (Throwable exception) {
byte[] errorBytes = NettyUtils.errorPacket(400,
MessageFormatter.format("something goes wrong with channel:{}, exception={}",
ctx.getChannel(),
ExceptionUtils.getStackTrace(exception)).getMessage());
NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
.getRemoteAddress()
.toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
} finally {
MDC.remove("destination");
}
}
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
logger.error("something goes wrong with channel:{}, exception={}",
ctx.getChannel(),
ExceptionUtils.getStackTrace(e.getCause()));
ctx.getChannel().close();
}
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
}
private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (runningMonitor.isStart()) {
runningMonitor.release();
}
}
}
private TimeUnit convertTimeUnit(int unit) {
switch (unit) {
case 0:
return TimeUnit.NANOSECONDS;
case 1:
return TimeUnit.MICROSECONDS;
case 2:
return TimeUnit.MILLISECONDS;
case 3:
return TimeUnit.SECONDS;
case 4:
return TimeUnit.MINUTES;
case 5:
return TimeUnit.HOURS;
case 6:
return TimeUnit.DAYS;
default:
return TimeUnit.MILLISECONDS;
}
}
public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embeddedServer = embeddedServer;
}
}
|
