1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.jboss.netty.example.localtime;
17
18 import java.util.ArrayList;
19 import java.util.Collection;
20 import java.util.Formatter;
21 import java.util.List;
22 import java.util.concurrent.BlockingQueue;
23 import java.util.concurrent.LinkedBlockingQueue;
24 import java.util.logging.Level;
25 import java.util.logging.Logger;
26
27 import org.jboss.netty.channel.Channel;
28 import org.jboss.netty.channel.ChannelEvent;
29 import org.jboss.netty.channel.ChannelHandlerContext;
30 import org.jboss.netty.channel.ChannelStateEvent;
31 import org.jboss.netty.channel.ExceptionEvent;
32 import org.jboss.netty.channel.MessageEvent;
33 import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
34 import org.jboss.netty.example.localtime.LocalTimeProtocol.Continent;
35 import org.jboss.netty.example.localtime.LocalTimeProtocol.LocalTime;
36 import org.jboss.netty.example.localtime.LocalTimeProtocol.LocalTimes;
37 import org.jboss.netty.example.localtime.LocalTimeProtocol.Location;
38 import org.jboss.netty.example.localtime.LocalTimeProtocol.Locations;
39
40 public class LocalTimeClientHandler extends SimpleChannelUpstreamHandler {
41
42 private static final Logger logger = Logger.getLogger(
43 LocalTimeClientHandler.class.getName());
44
45
46 private volatile Channel channel;
47 private final BlockingQueue<LocalTimes> answer = new LinkedBlockingQueue<LocalTimes>();
48
49 public List<String> getLocalTimes(Collection<String> cities) {
50 Locations.Builder builder = Locations.newBuilder();
51
52 for (String c: cities) {
53 String[] components = c.split("/");
54 builder.addLocation(Location.newBuilder().
55 setContinent(Continent.valueOf(components[0].toUpperCase())).
56 setCity(components[1]).build());
57 }
58
59 channel.write(builder.build());
60
61 LocalTimes localTimes;
62 boolean interrupted = false;
63 for (;;) {
64 try {
65 localTimes = answer.take();
66 break;
67 } catch (InterruptedException e) {
68 interrupted = true;
69 }
70 }
71
72 if (interrupted) {
73 Thread.currentThread().interrupt();
74 }
75
76 List<String> result = new ArrayList<String>();
77 for (LocalTime lt: localTimes.getLocalTimeList()) {
78 result.add(
79 new Formatter().format(
80 "%4d-%02d-%02d %02d:%02d:%02d %s",
81 lt.getYear(),
82 lt.getMonth(),
83 lt.getDayOfMonth(),
84 lt.getHour(),
85 lt.getMinute(),
86 lt.getSecond(),
87 lt.getDayOfWeek().name()).toString());
88 }
89
90 return result;
91 }
92
93 @Override
94 public void handleUpstream(
95 ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
96 if (e instanceof ChannelStateEvent) {
97 logger.info(e.toString());
98 }
99 super.handleUpstream(ctx, e);
100 }
101
102 @Override
103 public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
104 throws Exception {
105 channel = e.getChannel();
106 super.channelOpen(ctx, e);
107 }
108
109 @Override
110 public void messageReceived(
111 ChannelHandlerContext ctx, final MessageEvent e) {
112 boolean offered = answer.offer((LocalTimes) e.getMessage());
113 assert offered;
114 }
115
116 @Override
117 public void exceptionCaught(
118 ChannelHandlerContext ctx, ExceptionEvent e) {
119 logger.log(
120 Level.WARNING,
121 "Unexpected exception from downstream.",
122 e.getCause());
123 e.getChannel().close();
124 }
125 }