1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.handler.stream;
21
22 import java.io.IOException;
23 import java.io.InputStream;
24 import java.io.OutputStream;
25 import java.net.SocketTimeoutException;
26
27 import org.apache.mina.core.buffer.IoBuffer;
28 import org.apache.mina.core.service.IoHandler;
29 import org.apache.mina.core.service.IoHandlerAdapter;
30 import org.apache.mina.core.session.AttributeKey;
31 import org.apache.mina.core.session.IdleStatus;
32 import org.apache.mina.core.session.IoSession;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
35
36
37
38
39
40
41
42
43
44
45
46 public abstract class StreamIoHandler extends IoHandlerAdapter {
47 private final static Logger LOGGER = LoggerFactory.getLogger(StreamIoHandler.class);
48
49 private static final AttributeKey KEY_IN = new AttributeKey(StreamIoHandler.class, "in");
50 private static final AttributeKey KEY_OUT = new AttributeKey(StreamIoHandler.class, "out");
51
52 private int readTimeout;
53
54 private int writeTimeout;
55
56 protected StreamIoHandler() {
57
58 }
59
60
61
62
63
64
65 protected abstract void processStreamIo(IoSession session, InputStream in,
66 OutputStream out);
67
68
69
70
71
72 public int getReadTimeout() {
73 return readTimeout;
74 }
75
76
77
78
79
80 public void setReadTimeout(int readTimeout) {
81 this.readTimeout = readTimeout;
82 }
83
84
85
86
87
88 public int getWriteTimeout() {
89 return writeTimeout;
90 }
91
92
93
94
95
96 public void setWriteTimeout(int writeTimeout) {
97 this.writeTimeout = writeTimeout;
98 }
99
100
101
102
103 @Override
104 public void sessionOpened(IoSession session) {
105
106 session.getConfig().setWriteTimeout(writeTimeout);
107 session.getConfig().setIdleTime(IdleStatus.READER_IDLE, readTimeout);
108
109
110 InputStream in = new IoSessionInputStream();
111 OutputStream out = new IoSessionOutputStream(session);
112 session.setAttribute(KEY_IN, in);
113 session.setAttribute(KEY_OUT, out);
114 processStreamIo(session, in, out);
115 }
116
117
118
119
120 @Override
121 public void sessionClosed(IoSession session) throws Exception {
122 final InputStream in = (InputStream) session.getAttribute(KEY_IN);
123 final OutputStream out = (OutputStream) session.getAttribute(KEY_OUT);
124 try {
125 in.close();
126 } finally {
127 out.close();
128 }
129 }
130
131
132
133
134 @Override
135 public void messageReceived(IoSession session, Object buf) {
136 final IoSessionInputStream in = (IoSessionInputStream) session
137 .getAttribute(KEY_IN);
138 in.write((IoBuffer) buf);
139 }
140
141
142
143
144 @Override
145 public void exceptionCaught(IoSession session, Throwable cause) {
146 final IoSessionInputStream in = (IoSessionInputStream) session
147 .getAttribute(KEY_IN);
148
149 IOException e = null;
150 if (cause instanceof StreamIoException) {
151 e = (IOException) cause.getCause();
152 } else if (cause instanceof IOException) {
153 e = (IOException) cause;
154 }
155
156 if (e != null && in != null) {
157 in.throwException(e);
158 } else {
159 LOGGER.warn("Unexpected exception.", cause);
160 session.close(true);
161 }
162 }
163
164
165
166
167 @Override
168 public void sessionIdle(IoSession session, IdleStatus status) {
169 if (status == IdleStatus.READER_IDLE) {
170 throw new StreamIoException(new SocketTimeoutException(
171 "Read timeout"));
172 }
173 }
174
175 private static class StreamIoException extends RuntimeException {
176 private static final long serialVersionUID = 3976736960742503222L;
177
178 public StreamIoException(IOException cause) {
179 super(cause);
180 }
181 }
182 }