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