1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.common;
21
22 import java.util.ArrayList;
23 import java.util.Collection;
24 import java.util.Iterator;
25 import java.util.List;
26 import java.util.concurrent.TimeUnit;
27
28
29
30
31
32
33
34
35 public class IoUtil {
36
37 private static final IoSession[] EMPTY_SESSIONS = new IoSession[0];
38
39
40
41
42
43
44 public static List<WriteFuture> broadcast(Object message, Collection<IoSession> sessions) {
45 List<WriteFuture> answer = new ArrayList<WriteFuture>(sessions.size());
46 broadcast(message, sessions.iterator(), answer);
47 return answer;
48 }
49
50
51
52
53
54
55 public static List<WriteFuture> broadcast(Object message, Iterable<IoSession> sessions) {
56 List<WriteFuture> answer = new ArrayList<WriteFuture>();
57 broadcast(message, sessions.iterator(), answer);
58 return answer;
59 }
60
61
62
63
64
65
66 public static List<WriteFuture> broadcast(Object message, Iterator<IoSession> sessions) {
67 List<WriteFuture> answer = new ArrayList<WriteFuture>();
68 broadcast(message, sessions, answer);
69 return answer;
70 }
71
72
73
74
75
76
77 public static List<WriteFuture> broadcast(Object message, IoSession... sessions) {
78 if (sessions == null) {
79 sessions = EMPTY_SESSIONS;
80 }
81
82 List<WriteFuture> answer = new ArrayList<WriteFuture>(sessions.length);
83 if (message instanceof IoBuffer) {
84 for (IoSession s: sessions) {
85 answer.add(s.write(((IoBuffer) message).duplicate()));
86 }
87 } else {
88 for (IoSession s: sessions) {
89 answer.add(s.write(message));
90 }
91 }
92 return answer;
93 }
94
95 private static void broadcast(Object message, Iterator<IoSession> sessions, Collection<WriteFuture> answer) {
96 if (message instanceof IoBuffer) {
97 while (sessions.hasNext()) {
98 IoSession s = sessions.next();
99 answer.add(s.write(((IoBuffer) message).duplicate()));
100 }
101 } else {
102 while (sessions.hasNext()) {
103 IoSession s = sessions.next();
104 answer.add(s.write(message));
105 }
106 }
107 }
108
109 public static void await(Iterable<? extends IoFuture> futures) throws InterruptedException {
110 for (IoFuture f: futures) {
111 f.await();
112 }
113 }
114
115 public static void awaitUninterruptably(Iterable<? extends IoFuture> futures) {
116 for (IoFuture f: futures) {
117 f.awaitUninterruptibly();
118 }
119 }
120
121 public static boolean await(Iterable<? extends IoFuture> futures, long timeout, TimeUnit unit) throws InterruptedException {
122 return await(futures, unit.toMillis(timeout));
123 }
124
125 public static boolean await(Iterable<? extends IoFuture> futures, long timeoutMillis) throws InterruptedException {
126 return await0(futures, timeoutMillis, true);
127 }
128
129 public static boolean awaitUninterruptibly(Iterable<? extends IoFuture> futures, long timeout, TimeUnit unit) {
130 return awaitUninterruptibly(futures, unit.toMillis(timeout));
131 }
132
133 public static boolean awaitUninterruptibly(Iterable<? extends IoFuture> futures, long timeoutMillis) {
134 try {
135 return await0(futures, timeoutMillis, false);
136 } catch (InterruptedException e) {
137 throw new InternalError();
138 }
139 }
140
141 private static boolean await0(Iterable<? extends IoFuture> futures, long timeoutMillis, boolean interruptable) throws InterruptedException {
142 long startTime = timeoutMillis <= 0 ? 0 : System.currentTimeMillis();
143 long waitTime = timeoutMillis;
144
145 boolean lastComplete = true;
146 Iterator<? extends IoFuture> i = futures.iterator();
147 while (i.hasNext()) {
148 IoFuture f = i.next();
149 do {
150 if (interruptable) {
151 lastComplete = f.await(waitTime);
152 } else {
153 lastComplete = f.awaitUninterruptibly(waitTime);
154 }
155
156 waitTime = timeoutMillis - (System.currentTimeMillis() - startTime);
157
158 if (lastComplete || waitTime <= 0) {
159 break;
160 }
161 } while (!lastComplete);
162
163 if (waitTime <= 0) {
164 break;
165 }
166 }
167
168 return lastComplete && !i.hasNext();
169 }
170
171 private IoUtil() {
172 }
173 }