#2875

pending completion

Build # #2875

Build Type

Pull #563

github-actions

Committed by

web-flow

Commit Message

Merge 430795bb9 into 19b20fefe

Pull Request Pull Request #563: Bump pgpainless

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

16372 of 41845 relevant lines covered (39.13%)

0.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.25

/smack-core/src/main/java/org/jivesoftware/smack/AsyncButOrdered.java

/**
 *
 * Copyright 2018-2019 Florian Schmaus
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.jivesoftware.smack;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.WeakHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;

/**
 * Helper class to perform an operation asynchronous but keeping the order in respect to a given key.
 * <p>
 * A typical use pattern for this helper class consists of callbacks for an abstract entity where the order of callbacks
 * matters, which eventually call user code in form of listeners. Since the order the callbacks matters, you need to use
 * synchronous connection listeners. But if those listeners would invoke the user provided listeners, and if those user
 * provided listeners would take a long time to complete, or even worse, block, then Smack's total progress is stalled,
 * since synchronous connection listeners are invoked from the main event loop.
 * </p>
 * <p>
 * It is common for those situations that the order of callbacks is not globally important, but only important in
 * respect to an particular entity. Take chat state notifications (CSN) for example: Assume there are two contacts which
 * send you CSNs. If a contact sends you first 'active' and then 'inactive, it is crucial that first the listener is
 * called with 'active' and afterwards with 'inactive'. But if there is another contact is sending 'composing' followed
 * by 'paused', then it is also important that the listeners are invoked in the correct order, but the order in which
 * the listeners for those two contacts are invoked does not matter.
 * </p>
 * <p>
 * Using this helper class, one would call {@link #performAsyncButOrdered(Object, Runnable)} which the remote contacts
 * JID as first argument and a {@link Runnable} invoking the user listeners as second. This class guarantees that
 * runnables of subsequent invocations are always executed after the runnables of previous invocations using the same
 * key.
 * </p>
 *
 * @param <K> the type of the key
 * @since 4.3
 */
public class AsyncButOrdered<K> {

    /**
     * A map with the currently pending runnables for a given key. Note that this is a weak hash map so we do not have
     * to take care of removing the keys ourselfs from the map.
     */
    private final Map<K, Queue<Runnable>> pendingRunnables = new WeakHashMap<>();

    /**
     * A marker map if there is an active thread for the given key. Holds the responsible handler thread if one is
     * active, otherwise the key is non-existend in the map.
     */
    private final Map<K, Handler> threadActiveMap = new HashMap<>();

    private final Executor executor;

    public AsyncButOrdered() {
        this(null);
    }

    public AsyncButOrdered(Executor executor) {
        this.executor = executor;
    }

    private void scheduleHandler(Handler handler) {
        if (executor == null) {
            AbstractXMPPConnection.asyncGo(handler);
        } else {
            executor.execute(handler);
        }
    }

    /**
     * Invoke the given {@link Runnable} asynchronous but ordered in respect to the given key.
     *
     * @param key the key deriving the order
     * @param runnable the {@link Runnable} to run
     * @return true if a new thread was created
     */
    public boolean performAsyncButOrdered(K key, Runnable runnable) {
        // First check if a key queue already exists, create one if not.
        Queue<Runnable> keyQueue;
        synchronized (pendingRunnables) {
            keyQueue = pendingRunnables.get(key);
            if (keyQueue == null) {
                keyQueue = new ConcurrentLinkedQueue<>();
                pendingRunnables.put(key, keyQueue);
            }
        }

        // Then add the task to the queue.
        keyQueue.add(runnable);

        // Finally check if there is already a handler working on that queue, create one if not.
        Handler newlyCreatedHandler = null;
        synchronized (threadActiveMap) {
            if (!threadActiveMap.containsKey(key)) {
                newlyCreatedHandler = new Handler(keyQueue, key);

                // Mark that there is thread active for the given key. Note that this has to be done before scheduling
                // the handler thread.
                threadActiveMap.put(key, newlyCreatedHandler);
            }
        }

        if (newlyCreatedHandler != null) {
            scheduleHandler(newlyCreatedHandler);
            return true;
        }

        return false;
    }

    public Executor asExecutorFor(final K key) {
        return new Executor() {
            @Override
            public void execute(Runnable runnable) {
                performAsyncButOrdered(key, runnable);
            }
        };
    }

    private class Handler implements Runnable {
        private final Queue<Runnable> keyQueue;
        private final K key;

        Handler(Queue<Runnable> keyQueue, K key) {
            this.keyQueue = keyQueue;
            this.key = key;
        }

        @Override
        public void run() {
            mainloop:
            while (true) {
                Runnable runnable = null;
                while ((runnable = keyQueue.poll()) != null) {
                    try {
                        runnable.run();
                    } catch (Throwable t) {
                        // The run() method threw, this handler thread is going to terminate because of that. We create
                        // a new handler to continue working on the queue while throwing the throwable so that the
                        // executor can handle it.
                        Handler newlyCreatedHandler = new Handler(keyQueue, key);
                        synchronized (threadActiveMap) {
                            threadActiveMap.put(key, newlyCreatedHandler);
                        }
                        scheduleHandler(newlyCreatedHandler);
                        throw t;
                    }
                }

                synchronized (threadActiveMap) {
                    // If the queue is empty, stop this handler, otherwise continue looping.
                    if (keyQueue.isEmpty()) {
                        threadActiveMap.remove(key);
                        break mainloop;
                    }
                }
            }
        }
    }
}

1	/**
2	*
3	* Copyright 2018-2019 Florian Schmaus
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License");
6	* you may not use this file except in compliance with the License.
7	* You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.jivesoftware.smack;
18
19	import java.util.HashMap;
20	import java.util.Map;
21	import java.util.Queue;
22	import java.util.WeakHashMap;
23	import java.util.concurrent.ConcurrentLinkedQueue;
24	import java.util.concurrent.Executor;
25
26	/**
27	* Helper class to perform an operation asynchronous but keeping the order in respect to a given key.
28	* <p>
29	* A typical use pattern for this helper class consists of callbacks for an abstract entity where the order of callbacks
30	* matters, which eventually call user code in form of listeners. Since the order the callbacks matters, you need to use
31	* synchronous connection listeners. But if those listeners would invoke the user provided listeners, and if those user
32	* provided listeners would take a long time to complete, or even worse, block, then Smack's total progress is stalled,
33	* since synchronous connection listeners are invoked from the main event loop.
34	* </p>
35	* <p>
36	* It is common for those situations that the order of callbacks is not globally important, but only important in
37	* respect to an particular entity. Take chat state notifications (CSN) for example: Assume there are two contacts which
38	* send you CSNs. If a contact sends you first 'active' and then 'inactive, it is crucial that first the listener is
39	* called with 'active' and afterwards with 'inactive'. But if there is another contact is sending 'composing' followed
40	* by 'paused', then it is also important that the listeners are invoked in the correct order, but the order in which
41	* the listeners for those two contacts are invoked does not matter.
42	* </p>
43	* <p>
44	* Using this helper class, one would call {@link #performAsyncButOrdered(Object, Runnable)} which the remote contacts
45	* JID as first argument and a {@link Runnable} invoking the user listeners as second. This class guarantees that
46	* runnables of subsequent invocations are always executed after the runnables of previous invocations using the same
47	* key.
48	* </p>
49	*
50	* @param <K> the type of the key
51	* @since 4.3
52	*/
53	public class AsyncButOrdered<K> {
54
55	/**
56	* A map with the currently pending runnables for a given key. Note that this is a weak hash map so we do not have
57	* to take care of removing the keys ourselfs from the map.
58	*/
59	private final Map<K, Queue<Runnable>> pendingRunnables = new WeakHashMap<>();	1✔
60
61	/**
62	* A marker map if there is an active thread for the given key. Holds the responsible handler thread if one is
63	* active, otherwise the key is non-existend in the map.
64	*/
65	private final Map<K, Handler> threadActiveMap = new HashMap<>();	1✔
66
67	private final Executor executor;
68
69	public AsyncButOrdered() {
70	this(null);	1✔
71	}	1✔
72
73	public AsyncButOrdered(Executor executor) {	1✔
74	this.executor = executor;	1✔
75	}	1✔
76
77	private void scheduleHandler(Handler handler) {
78	if (executor == null) {	1✔
79	AbstractXMPPConnection.asyncGo(handler);	1✔
80	} else {
81	executor.execute(handler);	×
82	}
83	}	1✔
84
85	/**
86	* Invoke the given {@link Runnable} asynchronous but ordered in respect to the given key.
87	*
88	* @param key the key deriving the order
89	* @param runnable the {@link Runnable} to run
90	* @return true if a new thread was created
91	*/
92	public boolean performAsyncButOrdered(K key, Runnable runnable) {
93	// First check if a key queue already exists, create one if not.
94	Queue<Runnable> keyQueue;
95	synchronized (pendingRunnables) {	1✔
96	keyQueue = pendingRunnables.get(key);	1✔
97	if (keyQueue == null) {	1✔
98	keyQueue = new ConcurrentLinkedQueue<>();	1✔
99	pendingRunnables.put(key, keyQueue);	1✔
100	}
101	}	1✔
102
103	// Then add the task to the queue.
104	keyQueue.add(runnable);	1✔
105
106	// Finally check if there is already a handler working on that queue, create one if not.
107	Handler newlyCreatedHandler = null;	1✔
108	synchronized (threadActiveMap) {	1✔
109	if (!threadActiveMap.containsKey(key)) {	1✔
110	newlyCreatedHandler = new Handler(keyQueue, key);	1✔
111
112	// Mark that there is thread active for the given key. Note that this has to be done before scheduling
113	// the handler thread.
114	threadActiveMap.put(key, newlyCreatedHandler);	1✔
115	}
116	}	1✔
117
118	if (newlyCreatedHandler != null) {	1✔
119	scheduleHandler(newlyCreatedHandler);	1✔
120	return true;	1✔
121	}
122
123	return false;	×
124	}
125
126	public Executor asExecutorFor(final K key) {
127	return new Executor() {	1✔
128	@Override
129	public void execute(Runnable runnable) {
130	performAsyncButOrdered(key, runnable);	1✔
131	}	1✔
132	};
133	}
134
135	private class Handler implements Runnable {
136	private final Queue<Runnable> keyQueue;
137	private final K key;
138
139	Handler(Queue<Runnable> keyQueue, K key) {	1✔
140	this.keyQueue = keyQueue;	1✔
141	this.key = key;	1✔
142	}	1✔
143
144	@Override
145	public void run() {
146	mainloop:
147	while (true) {
148	Runnable runnable = null;	1✔
149	while ((runnable = keyQueue.poll()) != null) {	1✔
150	try {
151	runnable.run();	1✔
152	} catch (Throwable t) {	×
153	// The run() method threw, this handler thread is going to terminate because of that. We create
154	// a new handler to continue working on the queue while throwing the throwable so that the
155	// executor can handle it.
156	Handler newlyCreatedHandler = new Handler(keyQueue, key);	×
157	synchronized (threadActiveMap) {	×
158	threadActiveMap.put(key, newlyCreatedHandler);	×
159	}	×
160	scheduleHandler(newlyCreatedHandler);	×
161	throw t;	×
162	}	1✔
163	}
164
165	synchronized (threadActiveMap) {	1✔
166	// If the queue is empty, stop this handler, otherwise continue looping.
167	if (keyQueue.isEmpty()) {	1✔
168	threadActiveMap.remove(key);	1✔
169	break mainloop;	1✔
170	}
171	}	×
172	}	×
173	}	1✔
174	}
175	}

igniterealtime / Smack / #2875

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous