Skip to content

Commit

Permalink
[DROOLS-1111] make listeners registration on stateless session thread…
Browse files Browse the repository at this point in the history 
… safe
  • Loading branch information
@mariofusco
mariofusco committed Apr 1, 2016
1 parent 1e50f13 commit 01b8fc5
Showing 1 changed file with 71 additions and 110 deletions.
181 changes: 71 additions & 110 deletions drools-core/src/main/java/org/drools/core/impl/StatelessKnowledgeSessionImpl.java
View file
Expand Up @@ -26,11 +26,7 @@
import org.drools.core.command.runtime.rule.FireAllRulesCommand;
import org.drools.core.common.InternalFactHandle;
import org.drools.core.common.WorkingMemoryFactory;
import org.drools.core.event.AgendaEventSupport;
import org.drools.core.event.ProcessEventSupport;
import org.drools.core.event.RuleRuntimeEventSupport;
import org.drools.core.runtime.impl.ExecutionResultImpl;
import org.drools.core.runtime.process.InternalProcessRuntime;
import org.kie.api.KieBase;
import org.kie.api.command.Command;
import org.kie.api.event.process.ProcessEventListener;
Expand All @@ -51,11 +47,11 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EventListener;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;

public class StatelessKnowledgeSessionImpl extends AbstractRuntime
implements
Expand All @@ -67,15 +63,7 @@ public class StatelessKnowledgeSessionImpl extends AbstractRuntime
private MapGlobalResolver sessionGlobals = new MapGlobalResolver();
private Map<String, Channel> channels = new HashMap<String, Channel>();

/** The event mapping */
public Set<RuleRuntimeEventListener> cachedRuleRuntimeListeners;
public Set<AgendaEventListener> cachedAgendaListeners;
public Set<ProcessEventListener> cachedProcessEventListener;

/** The event support */
private AgendaEventSupport agendaEventSupport = new AgendaEventSupport();
private RuleRuntimeEventSupport ruleRuntimeEventSupport = new RuleRuntimeEventSupport();
private ProcessEventSupport processEventSupport = new ProcessEventSupport();
private final List<ListnerHolder> listeners = new CopyOnWriteArrayList<ListnerHolder>();

private KieSessionConfiguration conf;
private Environment environment;
Expand Down Expand Up @@ -120,21 +108,9 @@ public StatefulKnowledgeSession newWorkingMemory() {
this.environment);
StatefulKnowledgeSessionImpl ksessionImpl = (StatefulKnowledgeSessionImpl) ksession;

// we don't pass the mapped listener wrappers to the session constructor anymore,
// because they would be ignored anyway, since the wm already contains those listeners

((Globals) ksessionImpl.getGlobalResolver()).setDelegate(this.sessionGlobals);

// copy over the default generated listeners that are used for internal stuff once
registerSystemListeners(ksessionImpl);
registerCustomListeners();

ksessionImpl.setAgendaEventSupport( this.agendaEventSupport );
ksessionImpl.setRuleRuntimeEventSupport(this.ruleRuntimeEventSupport);
InternalProcessRuntime processRuntime = ksessionImpl.internalGetProcessRuntime();
if ( processRuntime != null ) {
processRuntime.setProcessEventSupport( this.processEventSupport );
}
registerListeners( ksessionImpl );

for( Map.Entry<String, Channel> entry : this.channels.entrySet() ) {
ksession.registerChannel( entry.getKey(), entry.getValue() );
Expand All @@ -146,109 +122,76 @@ public StatefulKnowledgeSession newWorkingMemory() {
}
}

private void registerSystemListeners(StatefulKnowledgeSessionImpl wm) {
for (AgendaEventListener listener : wm.getAgendaEventSupport().getEventListeners()) {
this.agendaEventSupport.addEventListener(listener);
}
for (RuleRuntimeEventListener listener : wm.getRuleRuntimeEventSupport().getEventListeners()) {
this.ruleRuntimeEventSupport.addEventListener(listener);
private void registerListeners( StatefulKnowledgeSessionImpl wm ) {
if ( listeners.isEmpty()) {
return;
}
InternalProcessRuntime processRuntime = wm.internalGetProcessRuntime();
if ( processRuntime != null ) {
for ( ProcessEventListener listener : processRuntime.getProcessEventListeners() ) {
this.processEventSupport.addEventListener( listener );
}
}
}

private void registerCustomListeners() {
if ( cachedAgendaListeners != null ) {
for (AgendaEventListener agendaListener : cachedAgendaListeners) {
this.agendaEventSupport.addEventListener( agendaListener );
}
}
if ( cachedRuleRuntimeListeners != null ) {
for (RuleRuntimeEventListener wmListener : cachedRuleRuntimeListeners) {
this.ruleRuntimeEventSupport.addEventListener(wmListener);
}
}
if ( cachedProcessEventListener != null ) {
for (ProcessEventListener processListener : cachedProcessEventListener) {
this.processEventSupport.addEventListener( processListener );
for (ListnerHolder listnerHolder : listeners ) {
switch (listnerHolder.type) {
case AGENDA:
wm.addEventListener( (AgendaEventListener)listnerHolder.listener );
break;
case RUNTIME:
wm.addEventListener( (RuleRuntimeEventListener)listnerHolder.listener );
break;
case PROCESS:
wm.addEventListener( (ProcessEventListener)listnerHolder.listener );
break;
}
}
}

public void addEventListener(AgendaEventListener listener) {
registerAgendaEventListener(listener);
}

private void registerAgendaEventListener(AgendaEventListener listener) {
if ( this.cachedAgendaListeners == null ) {
this.cachedAgendaListeners = new HashSet<AgendaEventListener>();
}
this.cachedAgendaListeners.add(listener);
listeners.add( new ListnerHolder( ListnerHolder.Type.AGENDA, listener ) );
}

public Collection<AgendaEventListener> getAgendaEventListeners() {
return cachedAgendaListeners != null ? Collections.unmodifiableCollection( cachedAgendaListeners ) : Collections.<AgendaEventListener>emptySet();
return (Collection<AgendaEventListener>) getListeners(ListnerHolder.Type.AGENDA);
}

public void removeEventListener(AgendaEventListener listener) {
if ( this.cachedAgendaListeners != null ) {
cachedAgendaListeners.remove( listener );
this.agendaEventSupport.removeEventListener( listener );
}
listeners.remove( new ListnerHolder( ListnerHolder.Type.AGENDA, listener ) );
}

public void addEventListener(RuleRuntimeEventListener listener) {
registerRuleRuntimeEventListener(listener);
}

private void registerRuleRuntimeEventListener(RuleRuntimeEventListener listener) {
if ( this.cachedRuleRuntimeListeners == null ) {
this.cachedRuleRuntimeListeners = new HashSet<RuleRuntimeEventListener>();
}
this.cachedRuleRuntimeListeners.add(listener);
listeners.add( new ListnerHolder( ListnerHolder.Type.RUNTIME, listener ) );
}

public void removeEventListener(RuleRuntimeEventListener listener) {
if ( this.cachedRuleRuntimeListeners != null ) {
this.ruleRuntimeEventSupport.removeEventListener(listener);
}
listeners.remove( new ListnerHolder( ListnerHolder.Type.RUNTIME, listener ) );
}

public Collection<RuleRuntimeEventListener> getRuleRuntimeEventListeners() {
if ( this.cachedRuleRuntimeListeners == null ) {
this.cachedRuleRuntimeListeners = new HashSet<RuleRuntimeEventListener>();
}

return Collections.unmodifiableCollection( this.cachedRuleRuntimeListeners );
return (Collection<RuleRuntimeEventListener>) getListeners(ListnerHolder.Type.RUNTIME);
}

public void addEventListener(ProcessEventListener listener) {
if ( this.cachedProcessEventListener == null ) {
this.cachedProcessEventListener = new HashSet<ProcessEventListener>();
}
this.cachedProcessEventListener.add(listener);
this.processEventSupport.addEventListener(listener);
listeners.add( new ListnerHolder( ListnerHolder.Type.PROCESS, listener ) );
}

public Collection<ProcessEventListener> getProcessEventListeners() {
return Collections.unmodifiableCollection( this.cachedProcessEventListener );
return (Collection<ProcessEventListener>) getListeners(ListnerHolder.Type.PROCESS);
}

public void removeEventListener(ProcessEventListener listener) {
if (this.cachedProcessEventListener != null) {
this.cachedProcessEventListener.remove(listener);
private Collection<? extends EventListener> getListeners(ListnerHolder.Type type) {
if ( listeners.isEmpty()) {
return Collections.emptySet();
}
Collection<EventListener> l = new ArrayList<EventListener>();
for (ListnerHolder listnerHolder : listeners ) {
if (listnerHolder.type == type) {
l.add( listnerHolder.listener );
}
}
this.processEventSupport.removeEventListener( listener );
return l;
}

public void setGlobal(String identifier,
Object value) {
this.sessionGlobals.setGlobal(identifier,
value);
public void removeEventListener(ProcessEventListener listener) {
listeners.remove( new ListnerHolder( ListnerHolder.Type.RUNTIME, listener ) );
}

public void setGlobal(String identifier, Object value) {
this.sessionGlobals.setGlobal(identifier, value);
}

public Globals getGlobals() {
Expand Down Expand Up @@ -361,20 +304,38 @@ public Environment getEnvironment() {
}

protected void dispose(StatefulKnowledgeSession ksession) {
StatefulKnowledgeSessionImpl wm = ((StatefulKnowledgeSessionImpl) ksession);
ksession.dispose();
}

for ( AgendaEventListener listener : wm.getAgendaEventSupport().getEventListeners() ) {
this.agendaEventSupport.removeEventListener( listener );
}
for ( RuleRuntimeEventListener listener: wm.getRuleRuntimeEventSupport().getEventListeners() ) {
this.ruleRuntimeEventSupport.removeEventListener(listener);
private static class ListnerHolder {
enum Type { AGENDA, RUNTIME, PROCESS }

final Type type;
final EventListener listener;

private ListnerHolder( Type type, EventListener listener ) {
this.type = type;
this.listener = listener;
}
InternalProcessRuntime processRuntime = wm.internalGetProcessRuntime();
if ( processRuntime != null ) {
for ( ProcessEventListener listener: processRuntime.getProcessEventListeners() ) {
this.processEventSupport.removeEventListener( listener );

@Override
public boolean equals( Object obj ) {
if ( this == obj ) {
return true;
}
if ( obj == null || !(obj instanceof ListnerHolder) ) {
return false;
}

ListnerHolder that = (ListnerHolder) obj;
return type == that.type && listener == that.listener;
}

@Override
public int hashCode() {
int result = type.hashCode();
result = 31 * result + listener.hashCode();
return result;
}
ksession.dispose();
}
}

0 comments on commit 01b8fc5

Please sign in to comment.