You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@jackrabbit.apache.org by re...@apache.org on 2020/06/08 09:54:42 UTC
svn commit: r1878589 - in /jackrabbit/branches/2.18: ./
jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java
Author: reschke
Date: Mon Jun 8 09:54:41 2020
New Revision: 1878589
URL: http://svn.apache.org/viewvc?rev=1878589&view=rev
Log:
JCR-4530: jackrabbit-core: avoid use of deprecated commons-collections Buffers (merged r1873852 into 2.18)
Modified:
jackrabbit/branches/2.18/ (props changed)
jackrabbit/branches/2.18/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java
Propchange: jackrabbit/branches/2.18/
------------------------------------------------------------------------------
Merged /jackrabbit/trunk:r1873852
Modified: jackrabbit/branches/2.18/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java
URL: http://svn.apache.org/viewvc/jackrabbit/branches/2.18/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java?rev=1878589&r1=1878588&r2=1878589&view=diff
==============================================================================
--- jackrabbit/branches/2.18/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java (original)
+++ jackrabbit/branches/2.18/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/observation/ObservationDispatcher.java Mon Jun 8 09:54:41 2020
@@ -16,19 +16,17 @@
*/
package org.apache.jackrabbit.core.observation;
-import org.apache.commons.collections.Buffer;
-import org.apache.commons.collections.BufferUtils;
-import org.apache.commons.collections.buffer.UnboundedFifoBuffer;
-import org.apache.jackrabbit.core.state.ChangeLog;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.util.Collections;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.jackrabbit.core.state.ChangeLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* Dispatcher for dispatching events to listeners within a single workspace.
*/
@@ -84,8 +82,7 @@ public final class ObservationDispatcher
/**
* Contains the pending events that will be delivered to event listeners
*/
- private Buffer eventQueue
- = BufferUtils.blockingBuffer(new UnboundedFifoBuffer());
+ private BlockingQueue<DispatchAction> eventQueue = new LinkedBlockingQueue<>();
private AtomicInteger eventQueueSize = new AtomicInteger();
@@ -116,7 +113,7 @@ public final class ObservationDispatcher
try {
notificationThread.join();
} catch (InterruptedException e) {
- // FIXME log exception ?
+ log.debug("while joining notificationThread", e);
}
log.info("Notification of EventListeners stopped.");
}
@@ -149,26 +146,29 @@ public final class ObservationDispatcher
* thread.
*/
public void run() {
- DispatchAction action;
- while ((action = (DispatchAction) eventQueue.remove()) != DISPOSE_MARKER) {
-
- eventQueueSize.getAndAdd(-action.getEventStates().size());
- log.debug("got EventStateCollection");
- log.debug("event delivery to " + action.getEventConsumers().size() + " consumers started...");
- for (Iterator<EventConsumer> it = action.getEventConsumers().iterator(); it.hasNext();) {
- EventConsumer c = it.next();
- try {
- c.consumeEvents(action.getEventStates());
- } catch (Throwable t) {
- log.warn("EventConsumer " +
- c.getEventListener().getClass().getName() +
- " threw exception", t);
- // move on to the next consumer
+ boolean done = false;
+ do {
+ try {
+ DispatchAction action = eventQueue.take();
+ done = action == DISPOSE_MARKER;
+ if (!done) {
+ eventQueueSize.getAndAdd(-action.getEventStates().size());
+ log.debug("got EventStateCollection");
+ log.debug("event delivery to " + action.getEventConsumers().size() + " consumers started...");
+ for (EventConsumer c : action.getEventConsumers()) {
+ try {
+ c.consumeEvents(action.getEventStates());
+ } catch (Throwable t) {
+ log.warn("EventConsumer " + c.getEventListener().getClass().getName() + " threw exception", t);
+ // move on to the next consumer
+ }
+ }
}
+ } catch (InterruptedException ex) {
+ log.debug("event delivery interrupted", ex);
}
- log.debug("event delivery finished.");
-
- }
+ } while (!done);
+ log.debug("event delivery finished.");
}
/**
@@ -310,5 +310,4 @@ public final class ObservationDispatcher
}
}
}
-
}