1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.jboss.netty.handler.timeout;
17
18 import org.jboss.netty.bootstrap.ServerBootstrap;
19 import org.jboss.netty.channel.ChannelFuture;
20 import org.jboss.netty.channel.ChannelFutureListener;
21 import org.jboss.netty.channel.ChannelHandler.Sharable;
22 import org.jboss.netty.channel.ChannelHandlerContext;
23 import org.jboss.netty.channel.ChannelPipeline;
24 import org.jboss.netty.channel.ChannelPipelineFactory;
25 import org.jboss.netty.channel.Channels;
26 import org.jboss.netty.channel.MessageEvent;
27 import org.jboss.netty.channel.SimpleChannelDownstreamHandler;
28 import org.jboss.netty.util.ExternalResourceReleasable;
29 import org.jboss.netty.util.HashedWheelTimer;
30 import org.jboss.netty.util.Timeout;
31 import org.jboss.netty.util.Timer;
32 import org.jboss.netty.util.TimerTask;
33
34 import java.util.concurrent.TimeUnit;
35
36 import static org.jboss.netty.channel.Channels.*;
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75 @Sharable
76 public class WriteTimeoutHandler extends SimpleChannelDownstreamHandler
77 implements ExternalResourceReleasable {
78
79 static final WriteTimeoutException EXCEPTION = new WriteTimeoutException();
80
81 private final Timer timer;
82 private final long timeoutMillis;
83
84
85
86
87
88
89
90
91
92
93 public WriteTimeoutHandler(Timer timer, int timeoutSeconds) {
94 this(timer, timeoutSeconds, TimeUnit.SECONDS);
95 }
96
97
98
99
100
101
102
103
104
105
106
107
108 public WriteTimeoutHandler(Timer timer, long timeout, TimeUnit unit) {
109 if (timer == null) {
110 throw new NullPointerException("timer");
111 }
112 if (unit == null) {
113 throw new NullPointerException("unit");
114 }
115
116 this.timer = timer;
117 if (timeout <= 0) {
118 timeoutMillis = 0;
119 } else {
120 timeoutMillis = Math.max(unit.toMillis(timeout), 1);
121 }
122 }
123
124
125
126
127
128
129 public void releaseExternalResources() {
130 timer.stop();
131 }
132
133
134
135
136
137
138
139
140 protected long getTimeoutMillis(MessageEvent e) {
141 return timeoutMillis;
142 }
143
144 @Override
145 public void writeRequested(ChannelHandlerContext ctx, MessageEvent e)
146 throws Exception {
147
148 long timeoutMillis = getTimeoutMillis(e);
149 if (timeoutMillis > 0) {
150
151 ChannelFuture future = e.getFuture();
152 final Timeout timeout = timer.newTimeout(
153 new WriteTimeoutTask(ctx, future),
154 timeoutMillis, TimeUnit.MILLISECONDS);
155
156 future.addListener(new TimeoutCanceller(timeout));
157 }
158
159 super.writeRequested(ctx, e);
160 }
161
162 protected void writeTimedOut(ChannelHandlerContext ctx) throws Exception {
163 fireExceptionCaught(ctx, EXCEPTION);
164 }
165
166 private final class WriteTimeoutTask implements TimerTask {
167
168 private final ChannelHandlerContext ctx;
169 private final ChannelFuture future;
170
171 WriteTimeoutTask(ChannelHandlerContext ctx, ChannelFuture future) {
172 this.ctx = ctx;
173 this.future = future;
174 }
175
176 public void run(Timeout timeout) throws Exception {
177 if (timeout.isCancelled()) {
178 return;
179 }
180
181 if (!ctx.getChannel().isOpen()) {
182 return;
183 }
184
185
186 if (future.setFailure(EXCEPTION)) {
187
188 fireWriteTimeOut(ctx);
189 }
190 }
191
192 private void fireWriteTimeOut(final ChannelHandlerContext ctx) {
193 ctx.getPipeline().execute(new Runnable() {
194
195 public void run() {
196 try {
197 writeTimedOut(ctx);
198 } catch (Throwable t) {
199 fireExceptionCaught(ctx, t);
200 }
201 }
202 });
203 }
204 }
205
206 private static final class TimeoutCanceller implements ChannelFutureListener {
207 private final Timeout timeout;
208
209 TimeoutCanceller(Timeout timeout) {
210 this.timeout = timeout;
211 }
212
213 public void operationComplete(ChannelFuture future) throws Exception {
214 timeout.cancel();
215 }
216 }
217 }