Skip to contentPackage: IdleManager$2
IdleManager$2
| name | instruction | branch | complexity | line | method |
|---|
| run() |
|
|
|
|
|
| {...} |
|
|
|
|
|
Coverage
1: /*
2: * Copyright (c) 2014, 2023 Oracle and/or its affiliates. All rights reserved.
3: *
4: * This program and the accompanying materials are made available under the
5: * terms of the Eclipse Public License v. 2.0, which is available at
6: * http://www.eclipse.org/legal/epl-2.0.
7: *
8: * This Source Code may also be made available under the following Secondary
9: * Licenses when the conditions for such availability set forth in the
10: * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
11: * version 2 with the GNU Classpath Exception, which is available at
12: * https://www.gnu.org/software/classpath/license.html.
13: *
14: * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
15: */
16:
17: package org.eclipse.angus.mail.imap;
18:
19: import java.io.IOException;
20: import java.io.InterruptedIOException;
21: import java.net.Socket;
22: import java.nio.channels.*;
23: import java.util.*;
24: import java.util.logging.*;
25: import java.util.concurrent.ConcurrentLinkedQueue;
26: import java.util.concurrent.Executor;
27:
28: import jakarta.mail.*;
29:
30: import org.eclipse.angus.mail.util.MailLogger;
31:
32: /**
33: * IdleManager uses the optional IMAP IDLE command
34: * (<A HREF="http://www.ietf.org/rfc/rfc2177.txt">RFC 2177</A>)
35: * to watch multiple folders for new messages.
36: * IdleManager uses an Executor to execute tasks in separate threads.
37: * An Executor is typically provided by an ExecutorService.
38: * For example, for a Java SE application:
39: * <blockquote><pre>
40: * ExecutorService es = Executors.newCachedThreadPool();
41: * final IdleManager idleManager = new IdleManager(session, es);
42: * </pre></blockquote>
43: * For a Java EE 7 application:
44: * <blockquote><pre>
45: * {@literal @}Resource
46: * ManagedExecutorService es;
47: * final IdleManager idleManager = new IdleManager(session, es);
48: * </pre></blockquote>
49: * To watch for new messages in a folder, open the folder, register a listener,
50: * and ask the IdleManager to watch the folder:
51: * <blockquote><pre>
52: * Folder folder = store.getFolder("INBOX");
53: * folder.open(Folder.READ_WRITE);
54: * folder.addMessageCountListener(new MessageCountAdapter() {
55: * public void messagesAdded(MessageCountEvent ev) {
56: * Folder folder = (Folder)ev.getSource();
57: * Message[] msgs = ev.getMessages();
58: * System.out.println("Folder: " + folder +
59: * " got " + msgs.length + " new messages");
60: * try {
61: * // process new messages
62: * idleManager.watch(folder); // keep watching for new messages
63: * } catch (MessagingException mex) {
64: * // handle exception related to the Folder
65: * }
66: * }
67: * });
68: * idleManager.watch(folder);
69: * </pre></blockquote>
70: * This delivers the events for each folder in a separate thread, <b>NOT</b>
71: * using the Executor. To deliver all events in a single thread
72: * using the Executor, set the following properties for the Session
73: * (once), and then add listeners and watch the folder as above.
74: * <blockquote><pre>
75: * // the following should be done once...
76: * Properties props = session.getProperties();
77: * props.put("mail.event.scope", "session"); // or "application"
78: * props.put("mail.event.executor", es);
79: * </pre></blockquote>
80: * Note that, after processing new messages in your listener, or doing any
81: * other operations on the folder in any other thread, you need to tell
82: * the IdleManager to watch for more new messages. Unless, of course, you
83: * close the folder.
84: * <p>
85: * The IdleManager is created with a Session, which it uses only to control
86: * debug output. A single IdleManager instance can watch multiple Folders
87: * from multiple Stores and multiple Sessions.
88: * <p>
89: * Due to limitations in the Java SE nio support, a
90: * {@link java.nio.channels.SocketChannel SocketChannel} must be used instead
91: * of a {@link java.net.Socket Socket} to connect to the server. However,
92: * SocketChannels don't support all the features of Sockets, such as connecting
93: * through a SOCKS proxy server. SocketChannels also don't support
94: * simultaneous read and write, which means that the
95: * {@link IMAPFolder#idle idle} method can't be used if
96: * SocketChannels are being used; use this IdleManager instead.
97: * To enable support for SocketChannels instead of Sockets, set the
98: * <code>mail.imap.usesocketchannels</code> property in the Session used to
99: * access the IMAP Folder. (Or <code>mail.imaps.usesocketchannels</code> if
100: * you're using the "imaps" protocol.) This will effect all connections in
101: * that Session, but you can create another Session without this property set
102: * if you need to use the features that are incompatible with SocketChannels.
103: * <p>
104: * NOTE: The IdleManager, and all APIs and properties related to it, should
105: * be considered <strong>EXPERIMENTAL</strong>. They may be changed in the
106: * future in ways that are incompatible with applications using the
107: * current APIs.
108: *
109: * @since JavaMail 1.5.2
110: */
111: public class IdleManager {
112: private Executor es;
113: private Selector selector;
114: private MailLogger logger;
115: private volatile boolean die = false;
116: private volatile boolean running;
117: private Queue<IMAPFolder> toWatch = new ConcurrentLinkedQueue<>();
118: private Queue<IMAPFolder> toAbort = new ConcurrentLinkedQueue<>();
119:
120: /**
121: * Create an IdleManager. The Session is used only to configure
122: * debugging output. The Executor is used to create the
123: * "select" thread.
124: *
125: * @param session the Session containing configuration information
126: * @param es the Executor used to create threads
127: * @exception IOException for Selector failures
128: */
129: public IdleManager(Session session, Executor es) throws IOException {
130: this.es = es;
131: logger = new MailLogger(this.getClass(), "DEBUG IMAP",
132: session.getDebug(), session.getDebugOut());
133: selector = Selector.open();
134: es.execute(new Runnable() {
135: @Override
136: public void run() {
137: logger.fine("IdleManager select starting");
138: try {
139: running = true;
140: select();
141: } finally {
142: running = false;
143: logger.fine("IdleManager select terminating");
144: }
145: }
146: });
147: }
148:
149: /**
150: * Is the IdleManager currently running? The IdleManager starts
151: * running when the Executor schedules its task. The IdleManager
152: * stops running after its task detects the stop request from the
153: * {@link #stop stop} method, or if it terminates abnormally due
154: * to an unexpected error.
155: *
156: * @return true if the IdleMaanger is running
157: * @since JavaMail 1.5.5
158: */
159: public boolean isRunning() {
160: return running;
161: }
162:
163: /**
164: * Watch the Folder for new messages and other events using the IMAP IDLE
165: * command.
166: *
167: * @param folder the folder to watch
168: * @exception MessagingException for errors related to the folder
169: */
170: public void watch(Folder folder)
171: throws MessagingException {
172: if (die) // XXX - should be IllegalStateException?
173: throw new MessagingException("IdleManager is not running");
174: if (!(folder instanceof IMAPFolder))
175: throw new MessagingException("Can only watch IMAP folders");
176: IMAPFolder ifolder = (IMAPFolder)folder;
177: SocketChannel sc = ifolder.getChannel();
178: if (sc == null) {
179: if (folder.isOpen())
180: throw new MessagingException(
181: "Folder is not using SocketChannels");
182: else
183: throw new MessagingException("Folder is not open");
184: }
185: if (logger.isLoggable(Level.FINEST))
186: logger.log(Level.FINEST, "IdleManager watching {0}",
187: folderName(ifolder));
188: // keep trying to start the IDLE command until we're successful.
189: // may block if we're in the middle of aborting an IDLE command.
190: int tries = 0;
191: while (!ifolder.startIdle(this)) {
192: if (logger.isLoggable(Level.FINEST))
193: logger.log(Level.FINEST,
194: "IdleManager.watch startIdle failed for {0}",
195: folderName(ifolder));
196: tries++;
197: }
198: if (logger.isLoggable(Level.FINEST)) {
199: if (tries > 0)
200: logger.log(Level.FINEST,
201: "IdleManager.watch startIdle succeeded for {0}" +
202: " after " + tries + " tries",
203: folderName(ifolder));
204: else
205: logger.log(Level.FINEST,
206: "IdleManager.watch startIdle succeeded for {0}",
207: folderName(ifolder));
208: }
209: synchronized (this) {
210: toWatch.add(ifolder);
211: selector.wakeup();
212: }
213: }
214:
215: /**
216: * Request that the specified folder abort an IDLE command.
217: * We can't do the abort directly because the DONE message needs
218: * to be sent through the (potentially) SSL socket, which means
219: * we need to be in blocking I/O mode. We can only switch to
220: * blocking I/O mode when not selecting, so wake up the selector,
221: * which will process this request when it wakes up.
222: */
223: void requestAbort(IMAPFolder folder) {
224: toAbort.add(folder);
225: selector.wakeup();
226: }
227:
228: /**
229: * Run the {@link java.nio.channels.Selector#select select} loop
230: * to poll each watched folder for events sent from the server.
231: */
232: private void select() {
233: die = false;
234: try {
235: while (!die) {
236: watchAll();
237: logger.finest("IdleManager waiting...");
238: int ns = selector.select();
239: if (logger.isLoggable(Level.FINEST))
240: logger.log(Level.FINEST,
241: "IdleManager selected {0} channels", ns);
242: if (die || Thread.currentThread().isInterrupted())
243: break;
244:
245: /*
246: * Process any selected folders. We cancel the
247: * selection key for any selected folder, so if we
248: * need to continue watching that folder it's added
249: * to the toWatch list again. We can't actually
250: * register that folder again until the previous
251: * selection key is cancelled, so we call selectNow()
252: * just for the side effect of cancelling the selection
253: * keys. But if selectNow() selects something, we
254: * process it before adding folders from the toWatch
255: * queue. And so on until there is nothing to do, at
256: * which point it's safe to register folders from the
257: * toWatch queue. This should be "fair" since each
258: * selection key is used only once before being added
259: * to the toWatch list.
260: */
261: do {
262: processKeys();
263: } while (selector.selectNow() > 0 || !toAbort.isEmpty());
264: }
265: } catch (InterruptedIOException ex) {
266: logger.log(Level.FINEST, "IdleManager interrupted", ex);
267: } catch (IOException ex) {
268: logger.log(Level.FINEST, "IdleManager got I/O exception", ex);
269: } catch (Exception ex) {
270: logger.log(Level.FINEST, "IdleManager got exception", ex);
271: } finally {
272: die = true; // prevent new watches in case of exception
273: logger.finest("IdleManager unwatchAll");
274: try {
275: unwatchAll();
276: selector.close();
277: } catch (IOException ex2) {
278: // nothing to do...
279: logger.log(Level.FINEST, "IdleManager unwatch exception", ex2);
280: }
281: logger.fine("IdleManager exiting");
282: }
283: }
284:
285: /**
286: * Register all of the folders in the queue with the selector,
287: * switching them to nonblocking I/O mode first.
288: */
289: private void watchAll() {
290: /*
291: * Pull each of the folders from the toWatch queue
292: * and register it.
293: */
294: IMAPFolder folder;
295: while ((folder = toWatch.poll()) != null) {
296: if (logger.isLoggable(Level.FINEST))
297: logger.log(Level.FINEST,
298: "IdleManager adding {0} to selector", folderName(folder));
299: try {
300: SocketChannel sc = folder.getChannel();
301: if (sc == null)
302: continue;
303: // has to be non-blocking to select
304: sc.configureBlocking(false);
305: sc.register(selector, SelectionKey.OP_READ, folder);
306: } catch (IOException ex) {
307: // oh well, nothing to do
308: logger.log(Level.FINEST,
309: "IdleManager can't register folder", ex);
310: } catch (CancelledKeyException ex) {
311: // this should never happen
312: logger.log(Level.FINEST,
313: "IdleManager can't register folder", ex);
314: }
315: }
316: }
317:
318: /**
319: * Process the selected keys.
320: */
321: private void processKeys() throws IOException {
322: IMAPFolder folder;
323:
324: /*
325: * First, process any channels with data to read.
326: */
327: Set<SelectionKey> selectedKeys = selector.selectedKeys();
328: /*
329: * XXX - this is simpler, but it can fail with
330: * ConcurrentModificationException
331: *
332: for (SelectionKey sk : selectedKeys) {
333: selectedKeys.remove(sk); // only process each key once
334: ...
335: }
336: */
337: Iterator<SelectionKey> it = selectedKeys.iterator();
338: while (it.hasNext()) {
339: SelectionKey sk = it.next();
340: it.remove(); // only process each key once
341: // have to cancel so we can switch back to blocking I/O mode
342: sk.cancel();
343: folder = (IMAPFolder)sk.attachment();
344: if (logger.isLoggable(Level.FINEST))
345: logger.log(Level.FINEST,
346: "IdleManager selected folder: {0}", folderName(folder));
347: SelectableChannel sc = sk.channel();
348: // switch back to blocking to allow normal I/O
349: sc.configureBlocking(true);
350: try {
351: if (folder.handleIdle(false)) {
352: if (logger.isLoggable(Level.FINEST))
353: logger.log(Level.FINEST,
354: "IdleManager continue watching folder {0}",
355: folderName(folder));
356: // more to do with this folder, select on it again
357: toWatch.add(folder);
358: } else {
359: // done watching this folder,
360: if (logger.isLoggable(Level.FINEST))
361: logger.log(Level.FINEST,
362: "IdleManager done watching folder {0}",
363: folderName(folder));
364: }
365: } catch (MessagingException ex) {
366: // something went wrong, stop watching this folder
367: logger.log(Level.FINEST,
368: "IdleManager got exception for folder: " +
369: folderName(folder), ex);
370: }
371: }
372:
373: /*
374: * Now, process any folders that we need to abort.
375: */
376: while ((folder = toAbort.poll()) != null) {
377: if (logger.isLoggable(Level.FINEST))
378: logger.log(Level.FINEST,
379: "IdleManager aborting IDLE for folder: {0}",
380: folderName(folder));
381: SocketChannel sc = folder.getChannel();
382: if (sc == null)
383: continue;
384: SelectionKey sk = sc.keyFor(selector);
385: // have to cancel so we can switch back to blocking I/O mode
386: if (sk != null)
387: sk.cancel();
388: // switch back to blocking to allow normal I/O
389: sc.configureBlocking(true);
390:
391: // if there's a read timeout, have to do the abort in a new thread
392: Socket sock = sc.socket();
393: if (sock != null && sock.getSoTimeout() > 0) {
394: logger.finest("IdleManager requesting DONE with timeout");
395: toWatch.remove(folder);
396: final IMAPFolder folder0 = folder;
397: es.execute(new Runnable() {
398: @Override
399: public void run() {
400: // send the DONE and wait for the response
401: folder0.idleAbortWait();
402: }
403: });
404: } else {
405: folder.idleAbort(); // send the DONE message
406: // watch for OK response to DONE
407: // XXX - what if we also added it above? should be a nop
408: toWatch.add(folder);
409: }
410: }
411: }
412:
413: /**
414: * Stop watching all folders. Cancel any selection keys and,
415: * most importantly, switch the channel back to blocking mode.
416: * If there's any folders waiting to be watched, need to abort
417: * them too.
418: */
419: private void unwatchAll() {
420: IMAPFolder folder;
421: Set<SelectionKey> keys = selector.keys();
422: for (SelectionKey sk : keys) {
423: // have to cancel so we can switch back to blocking I/O mode
424: sk.cancel();
425: folder = (IMAPFolder)sk.attachment();
426: if (logger.isLoggable(Level.FINEST))
427: logger.log(Level.FINEST,
428: "IdleManager no longer watching folder: {0}",
429: folderName(folder));
430: SelectableChannel sc = sk.channel();
431: // switch back to blocking to allow normal I/O
432: try {
433: sc.configureBlocking(true);
434: folder.idleAbortWait(); // send the DONE message and wait
435: } catch (IOException ex) {
436: // ignore it, channel might be closed
437: logger.log(Level.FINEST,
438: "IdleManager exception while aborting idle for folder: " +
439: folderName(folder), ex);
440: }
441: }
442:
443: /*
444: * Finally, process any folders waiting to be watched.
445: */
446: while ((folder = toWatch.poll()) != null) {
447: if (logger.isLoggable(Level.FINEST))
448: logger.log(Level.FINEST,
449: "IdleManager aborting IDLE for unwatched folder: {0}",
450: folderName(folder));
451: SocketChannel sc = folder.getChannel();
452: if (sc == null)
453: continue;
454: try {
455: // channel should still be in blocking mode, but make sure
456: sc.configureBlocking(true);
457: folder.idleAbortWait(); // send the DONE message and wait
458: } catch (IOException ex) {
459: // ignore it, channel might be closed
460: logger.log(Level.FINEST,
461: "IdleManager exception while aborting idle for folder: " +
462: folderName(folder), ex);
463: }
464: }
465: }
466:
467: /**
468: * Stop the IdleManager. The IdleManager can not be restarted.
469: */
470: public synchronized void stop() {
471: die = true;
472: logger.fine("IdleManager stopping");
473: selector.wakeup();
474: }
475:
476: /**
477: * Return the fully qualified name of the folder, for use in log messages.
478: * Essentially just the getURLName method, but ignoring the
479: * MessagingException that can never happen.
480: */
481: private static String folderName(Folder folder) {
482: try {
483: return folder.getURLName().toString();
484: } catch (MessagingException mex) {
485: // can't happen
486: return folder.getStore().toString() + "/" + folder.toString();
487: }
488: }
489: }
![[Jenkins]](/angus/static/bf82f03a/images/svgs/logo.svg){kind=logo}
