13.5 编写监听者

    这一节我们编写一个监听者：EventLogMonitor，也就是用来接收数据的程序。EventLogMonitor做下面事情：

• 接收LogEventBroadcaster广播的DatagramPacket
• 解码LogEvent消息

• 输出LogEvent

  EventLogMonitor的示意图如下：
  

解码器代码如下：

[java] view plaincopy

1. package netty.in.action.udp;
3. import io.netty.buffer.ByteBuf;
4. import io.netty.channel.ChannelHandlerContext;
5. import io.netty.channel.socket.DatagramPacket;
6. import io.netty.handler.codec.MessageToMessageDecoder;
7. import io.netty.util.CharsetUtil;
9. import java.util.List;
11. public class LogEventDecoder extends MessageToMessageDecoder<DatagramPacket> {
13. @Override
14. protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List<Object> out)
15. throws Exception {
16. ByteBuf buf = msg.content();
17. int i = buf.indexOf(0, buf.readableBytes(), LogEvent.SEPARATOR);
18. String filename = buf.slice(0, i).toString(CharsetUtil.UTF_8);
19. String logMsg = buf.slice(i + 1, buf.readableBytes()).toString(CharsetUtil.UTF_8);
20. LogEvent event = new LogEvent(msg.sender(),
21. System.currentTimeMillis(), filename, logMsg);
22. out.add(event);
23. }
25. }

处理消息的Handler代码如下：

[java] view plaincopy

1. package netty.in.action.udp;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.channel.SimpleChannelInboundHandler;
6. public class LogEventHandler extends SimpleChannelInboundHandler<LogEvent> {
8. @Override
9. protected void channelRead0(ChannelHandlerContext ctx, LogEvent msg) throws Exception {
10. StringBuilder builder = new StringBuilder();
11. builder.append(msg.getReceived());
12. builder.append(" [");
13. builder.append(msg.getSource().toString());
14. builder.append("] [");
15. builder.append(msg.getLogfile());
16. builder.append("] : ");
17. builder.append(msg.getMsg());
18. System.out.println(builder.toString());
19. }
20. }

EventLogMonitor代码如下：

[java] view plaincopy

1. package netty.in.action.udp;
3. import io.netty.bootstrap.Bootstrap;
4. import io.netty.channel.Channel;
5. import io.netty.channel.ChannelInitializer;
6. import io.netty.channel.ChannelOption;
7. import io.netty.channel.ChannelPipeline;
8. import io.netty.channel.EventLoopGroup;
9. import io.netty.channel.nio.NioEventLoopGroup;
10. import io.netty.channel.socket.nio.NioDatagramChannel;
12. import java.net.InetSocketAddress;
14. public class LogEventMonitor {
16. private final EventLoopGroup group;
17. private final Bootstrap bootstrap;
19. public LogEventMonitor(InetSocketAddress address) {
20. group = new NioEventLoopGroup();
21. bootstrap = new Bootstrap();
22. bootstrap.group(group).channel(NioDatagramChannel.class)
23. .option(ChannelOption.SO_BROADCAST, true)
24. .handler(new ChannelInitializer<Channel>() {
25. @Override
26. protected void initChannel(Channel channel) throws Exception {
27. ChannelPipeline pipeline = channel.pipeline();
28. pipeline.addLast(new LogEventDecoder());
29. pipeline.addLast(new LogEventHandler());
30. }
31. }).localAddress(address);
32. }
34. public Channel bind() {
35. return bootstrap.bind().syncUninterruptibly().channel();
36. }
38. public void stop() {
39. group.shutdownGracefully();
40. }
42. public static void main(String[] args) throws InterruptedException {
44. LogEventMonitor monitor = new LogEventMonitor(new InetSocketAddress(4096));
45. try {
46. Channel channel = monitor.bind();
47. System.out.println("LogEventMonitor running");
48. channel.closeFuture().sync();
49. } finally {
50. monitor.stop();
51. }
52. }
53. }