1   /*
2    *  Licensed to the Apache Software Foundation (ASF) under one
3    *  or more contributor license agreements.  See the NOTICE file
4    *  distributed with this work for additional information
5    *  regarding copyright ownership.  The ASF licenses this file
6    *  to you under the Apache License, Version 2.0 (the
7    *  "License"); you may not use this file except in compliance
8    *  with the License.  You may obtain a copy of the License at
9    *
10   *    http://www.apache.org/licenses/LICENSE-2.0
11   *
12   *  Unless required by applicable law or agreed to in writing,
13   *  software distributed under the License is distributed on an
14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   *  KIND, either express or implied.  See the License for the
16   *  specific language governing permissions and limitations
17   *  under the License.
18   *
19   */
20  package org.apache.mina.filter.executor;
21  
22  import java.net.SocketAddress;
23  
24  import junit.framework.Assert;
25  import junit.framework.TestCase;
26  
27  import org.apache.mina.common.CloseFuture;
28  import org.apache.mina.common.IdleStatus;
29  import org.apache.mina.common.IoFilterChain;
30  import org.apache.mina.common.IoHandler;
31  import org.apache.mina.common.IoService;
32  import org.apache.mina.common.IoServiceConfig;
33  import org.apache.mina.common.IoSession;
34  import org.apache.mina.common.IoSessionConfig;
35  import org.apache.mina.common.TransportType;
36  import org.apache.mina.common.IoFilter.NextFilter;
37  import org.apache.mina.common.IoFilter.WriteRequest;
38  import org.apache.mina.common.support.BaseIoSession;
39  
40  import java.util.concurrent.ThreadPoolExecutor;
41  import java.util.concurrent.TimeUnit;
42  
43  public class ExecutorFilterRegressionTest extends TestCase {
44      private ExecutorFilter filter;
45  
46      public ExecutorFilterRegressionTest() {
47      }
48  
49      public void setUp() throws Exception {
50          filter = new ExecutorFilter();
51      }
52  
53      public void tearDown() throws Exception {
54          ((ThreadPoolExecutor) filter.getExecutor()).shutdown();
55          filter = null;
56      }
57  
58      public void testEventOrder() throws Throwable {
59          final EventOrderChecker nextFilter = new EventOrderChecker();
60          final EventOrderCounter[] sessions = new EventOrderCounter[] {
61                  new EventOrderCounter(), new EventOrderCounter(),
62                  new EventOrderCounter(), new EventOrderCounter(),
63                  new EventOrderCounter(), new EventOrderCounter(),
64                  new EventOrderCounter(), new EventOrderCounter(),
65                  new EventOrderCounter(), new EventOrderCounter(), };
66          final int loop = 1000000;
67          final int end = sessions.length - 1;
68          final ExecutorFilter filter = this.filter;
69          ((ThreadPoolExecutor) filter.getExecutor()).setKeepAliveTime(3,
70                  TimeUnit.SECONDS);
71  
72          for (int i = 0; i < loop; i++) {
73              Integer objI = new Integer(i);
74  
75              for (int j = end; j >= 0; j--) {
76                  filter.messageReceived(nextFilter, sessions[j], objI);
77              }
78  
79              if (nextFilter.throwable != null) {
80                  throw nextFilter.throwable;
81              }
82          }
83  
84          Thread.sleep(1000);
85  
86          for (int i = end; i >= 0; i--) {
87              Assert.assertEquals(loop - 1, sessions[i].lastCount.intValue());
88          }
89      }
90  
91      private static class EventOrderCounter extends BaseIoSession {
92          private Integer lastCount = null;
93  
94          public synchronized void setLastCount(Integer newCount) {
95              if (lastCount != null) {
96                  Assert.assertEquals(lastCount.intValue() + 1, newCount
97                          .intValue());
98              }
99  
100             lastCount = newCount;
101         }
102 
103         public IoHandler getHandler() {
104             return null;
105         }
106 
107         public IoFilterChain getFilterChain() {
108             return null;
109         }
110 
111         public CloseFuture close() {
112             return null;
113         }
114 
115         public TransportType getTransportType() {
116             return null;
117         }
118 
119         public SocketAddress getRemoteAddress() {
120             return null;
121         }
122 
123         public SocketAddress getLocalAddress() {
124             return null;
125         }
126 
127         public int getScheduledWriteRequests() {
128             return 0;
129         }
130 
131         protected void updateTrafficMask() {
132         }
133 
134         public boolean isClosing() {
135             return false;
136         }
137 
138         public IoService getService() {
139             return null;
140         }
141 
142         public IoServiceConfig getServiceConfig() {
143             return null;
144         }
145 
146         public IoSessionConfig getConfig() {
147             return null;
148         }
149 
150         public SocketAddress getServiceAddress() {
151             return null;
152         }
153 
154         public int getScheduledWriteBytes() {
155             return 0;
156         }
157     }
158 
159     private static class EventOrderChecker implements NextFilter {
160         private Throwable throwable;
161 
162         public void sessionOpened(IoSession session) {
163         }
164 
165         public void sessionClosed(IoSession session) {
166         }
167 
168         public void sessionIdle(IoSession session, IdleStatus status) {
169         }
170 
171         public void exceptionCaught(IoSession session, Throwable cause) {
172         }
173 
174         public void messageReceived(IoSession session, Object message) {
175             try {
176                 ((EventOrderCounter) session).setLastCount((Integer) message);
177             } catch (Throwable t) {
178                 if (this.throwable == null) {
179                     this.throwable = t;
180                 }
181             }
182         }
183 
184         public void messageSent(IoSession session, Object message) {
185         }
186 
187         public void filterWrite(IoSession session, WriteRequest writeRequest) {
188         }
189 
190         public void filterClose(IoSession session) {
191         }
192 
193         public void sessionCreated(IoSession session) {
194         }
195     }
196 
197     public static void main(String[] args) {
198         junit.textui.TestRunner.run(ExecutorFilterRegressionTest.class);
199     }
200 }