1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.filter.stream;
21
22 import java.io.IOException;
23 import java.util.Queue;
24 import java.util.concurrent.ConcurrentLinkedQueue;
25
26 import org.apache.mina.core.buffer.IoBuffer;
27 import org.apache.mina.core.filterchain.IoFilterAdapter;
28 import org.apache.mina.core.filterchain.IoFilterChain;
29 import org.apache.mina.core.session.AttributeKey;
30 import org.apache.mina.core.session.IoSession;
31 import org.apache.mina.core.write.DefaultWriteRequest;
32 import org.apache.mina.core.write.WriteRequest;
33
34
35
36
37
38
39 public abstract class AbstractStreamWriteFilter<T> extends IoFilterAdapter {
40
41
42
43 public static final int DEFAULT_STREAM_BUFFER_SIZE = 4096;
44
45
46
47
48 protected final AttributeKey CURRENT_STREAM = new AttributeKey(getClass(), "stream");
49
50 protected final AttributeKey WRITE_REQUEST_QUEUE = new AttributeKey(getClass(), "queue");
51
52 protected final AttributeKey CURRENT_WRITE_REQUEST = new AttributeKey(getClass(), "writeRequest");
53
54 private int writeBufferSize = DEFAULT_STREAM_BUFFER_SIZE;
55
56 @Override
57 public void onPreAdd(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
58 Class<? extends IoFilterAdapter> clazz = getClass();
59 if (parent.contains(clazz)) {
60 throw new IllegalStateException("Only one " + clazz.getName() + " is permitted.");
61 }
62 }
63
64 @Override
65 public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
66
67 if (session.getAttribute(CURRENT_STREAM) != null) {
68 Queue<WriteRequest> queue = getWriteRequestQueue(session);
69 if (queue == null) {
70 queue = new ConcurrentLinkedQueue<WriteRequest>();
71 session.setAttribute(WRITE_REQUEST_QUEUE, queue);
72 }
73 queue.add(writeRequest);
74 return;
75 }
76
77 Object message = writeRequest.getMessage();
78
79 if (getMessageClass().isInstance(message)) {
80
81 T stream = getMessageClass().cast(message);
82
83 IoBuffer buffer = getNextBuffer(stream);
84 if (buffer == null) {
85
86 writeRequest.getFuture().setWritten();
87 nextFilter.messageSent(session, writeRequest);
88 } else {
89 session.setAttribute(CURRENT_STREAM, message);
90 session.setAttribute(CURRENT_WRITE_REQUEST, writeRequest);
91
92 nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
93 }
94
95 } else {
96 nextFilter.filterWrite(session, writeRequest);
97 }
98 }
99
100 abstract protected Class<T> getMessageClass();
101
102 @SuppressWarnings("unchecked")
103 private Queue<WriteRequest> getWriteRequestQueue(IoSession session) {
104 return (Queue<WriteRequest>) session.getAttribute(WRITE_REQUEST_QUEUE);
105 }
106
107 @SuppressWarnings("unchecked")
108 private Queue<WriteRequest> removeWriteRequestQueue(IoSession session) {
109 return (Queue<WriteRequest>) session.removeAttribute(WRITE_REQUEST_QUEUE);
110 }
111
112 @Override
113 public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
114 T stream = getMessageClass().cast(session.getAttribute(CURRENT_STREAM));
115
116 if (stream == null) {
117 nextFilter.messageSent(session, writeRequest);
118 } else {
119 IoBuffer buffer = getNextBuffer(stream);
120
121 if (buffer == null) {
122
123 session.removeAttribute(CURRENT_STREAM);
124 WriteRequest currentWriteRequest = (WriteRequest) session.removeAttribute(CURRENT_WRITE_REQUEST);
125
126
127 Queue<WriteRequest> queue = removeWriteRequestQueue(session);
128 if (queue != null) {
129 WriteRequest wr = queue.poll();
130 while (wr != null) {
131 filterWrite(nextFilter, session, wr);
132 wr = queue.poll();
133 }
134 }
135
136 currentWriteRequest.getFuture().setWritten();
137 nextFilter.messageSent(session, currentWriteRequest);
138 } else {
139 nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
140 }
141 }
142 }
143
144
145
146
147
148 public int getWriteBufferSize() {
149 return writeBufferSize;
150 }
151
152
153
154
155
156
157
158
159 public void setWriteBufferSize(int writeBufferSize) {
160 if (writeBufferSize < 1) {
161 throw new IllegalArgumentException("writeBufferSize must be at least 1");
162 }
163 this.writeBufferSize = writeBufferSize;
164 }
165
166 abstract protected IoBuffer getNextBuffer(T message) throws IOException;
167 }