Allow events from other primaries to be dispatched This results in core stream events command to also return the events from other primaries. Change-Id: Idb8334f8876b217fd32ca8b10937c4eaaa925a03
diff --git a/src/main/java/com/googlesource/gerrit/plugins/events/CoreListener.java b/src/main/java/com/googlesource/gerrit/plugins/events/CoreListener.java deleted file mode 100644 index 3336309..0000000 --- a/src/main/java/com/googlesource/gerrit/plugins/events/CoreListener.java +++ /dev/null
@@ -1,101 +0,0 @@ -// Copyright (C) 2016 The Android Open Source Project -// -// 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 com.googlesource.gerrit.plugins.events; - -import com.google.common.base.Supplier; -import com.google.gerrit.entities.EntitiesAdapterFactory; -import com.google.gerrit.entities.Project; -import com.google.gerrit.extensions.annotations.PluginName; -import com.google.gerrit.extensions.registration.DynamicSet; -import com.google.gerrit.server.config.GerritServerConfigProvider; -import com.google.gerrit.server.config.PluginConfig; -import com.google.gerrit.server.events.Event; -import com.google.gerrit.server.events.EventListener; -import com.google.gerrit.server.events.ProjectNameKeyAdapter; -import com.google.gerrit.server.events.SupplierSerializer; -import com.google.gson.Gson; -import com.google.gson.GsonBuilder; -import com.google.inject.Inject; -import com.google.inject.Singleton; -import java.io.IOException; -import java.util.HashSet; -import java.util.Set; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -@Singleton -public class CoreListener implements EventListener { - private static Logger log = LoggerFactory.getLogger(CoreListener.class); - - protected static final String KEY_FILTER = "filter"; - protected static final String FILTER_TYPE_DROP = "DROP"; - protected static final String FILTER_ELEMENT_CLASSNAME = "classname"; - - protected static final Gson gson = - new GsonBuilder() - .registerTypeAdapter(Supplier.class, new SupplierSerializer()) - .registerTypeAdapter(Project.NameKey.class, new ProjectNameKeyAdapter()) - .registerTypeAdapterFactory(EntitiesAdapterFactory.create()) - .create(); - - protected final String pluginName; - protected final GerritServerConfigProvider gerritServerConfigProvider; - protected final DynamicSet<StreamEventListener> listeners; - protected final EventStore store; - protected Set<String> dropEventNames = new HashSet<>(); - - @Inject - protected CoreListener( - @PluginName String pluginName, - GerritServerConfigProvider gerritServerConfigProvider, - EventStore store, - DynamicSet<StreamEventListener> listeners) { - this.pluginName = pluginName; - this.gerritServerConfigProvider = gerritServerConfigProvider; - this.store = store; - this.listeners = listeners; - readAndParseCfg(); - } - - @Override - public void onEvent(Event event) { - if (dropEventNames.contains(event.getClass().getName())) { - return; - } - try { - store.add(gson.toJson(event)); - } catch (IOException e) { - log.error("Cannot add event to event store", e); - } - for (StreamEventListener l : listeners) { - l.onStreamEventUpdate(); - } - } - - protected void readAndParseCfg() { - PluginConfig cfg = - PluginConfig.createFromGerritConfig(pluginName, gerritServerConfigProvider.loadConfig()); - for (String filter : cfg.getStringList(KEY_FILTER)) { - String pieces[] = filter.split(" "); - if (pieces.length == 3 - && FILTER_TYPE_DROP.equals(pieces[0]) - && FILTER_ELEMENT_CLASSNAME.equals(pieces[1])) { - dropEventNames.add(pieces[2]); - } else { - log.error("Ignoring invalid filter: " + filter); - } - } - } -}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/events/FileSystemEventBroker.java b/src/main/java/com/googlesource/gerrit/plugins/events/FileSystemEventBroker.java new file mode 100644 index 0000000..ef73a36 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/events/FileSystemEventBroker.java
@@ -0,0 +1,157 @@ +// Copyright (C) 2022 The Android Open Source Project +// +// 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 com.googlesource.gerrit.plugins.events; + +import com.google.gerrit.entities.BranchNameKey; +import com.google.gerrit.entities.Change; +import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.annotations.PluginName; +import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.config.GerritServerConfigProvider; +import com.google.gerrit.server.config.PluginConfig; +import com.google.gerrit.server.events.ChangeEvent; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.EventBroker; +import com.google.gerrit.server.events.EventGson; +import com.google.gerrit.server.events.EventListener; +import com.google.gerrit.server.events.ProjectEvent; +import com.google.gerrit.server.events.RefEvent; +import com.google.gerrit.server.events.UserScopedEventListener; +import com.google.gerrit.server.notedb.ChangeNotes; +import com.google.gerrit.server.permissions.PermissionBackend; +import com.google.gerrit.server.permissions.PermissionBackendException; +import com.google.gerrit.server.plugincontext.PluginSetContext; +import com.google.gerrit.server.project.ProjectCache; +import com.google.gson.Gson; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Singleton +public class FileSystemEventBroker extends EventBroker { + private static final Logger log = LoggerFactory.getLogger(FileSystemEventBroker.class); + protected final EventStore store; + protected final Gson gson; + protected final DynamicSet<StreamEventListener> streamEventListeners; + protected volatile long lastSent; + protected static final String KEY_FILTER = "filter"; + protected static final String FILTER_TYPE_DROP = "DROP"; + protected static final String FILTER_ELEMENT_CLASSNAME = "classname"; + protected Set<String> dropEventNames = new HashSet<>(); + + @Inject + public FileSystemEventBroker( + PluginSetContext<UserScopedEventListener> listeners, + PluginSetContext<EventListener> unrestrictedListeners, + PermissionBackend permissionBackend, + ProjectCache projectCache, + ChangeNotes.Factory notesFactory, + @Nullable @GerritInstanceId String gerritInstanceId, + EventStore store, + @EventGson Gson gson, + DynamicSet<StreamEventListener> streamEventListeners, + GerritServerConfigProvider gerritServerConfigProvider, + @PluginName String pluginName) + throws IOException { + super( + listeners, + unrestrictedListeners, + permissionBackend, + projectCache, + notesFactory, + gerritInstanceId); + this.store = store; + this.gson = gson; + this.streamEventListeners = streamEventListeners; + lastSent = store.getHead(); + readAndParseCfg(pluginName, gerritServerConfigProvider); + } + + @Override + public void postEvent(Change change, ChangeEvent event) throws PermissionBackendException { + storeEvent(event); + sendAllPendingEvents(); + } + + @Override + public void postEvent(Project.NameKey projectName, ProjectEvent event) { + storeEvent(event); + try { + sendAllPendingEvents(); + } catch (PermissionBackendException e) { + log.error("Permission Exception while dispatching the event. Will be tried again.", e); + } + } + + @Override + protected void fireEvent(BranchNameKey branchName, RefEvent event) + throws PermissionBackendException { + storeEvent(event); + sendAllPendingEvents(); + } + + @Override + public void postEvent(Event event) throws PermissionBackendException { + storeEvent(event); + sendAllPendingEvents(); + } + + protected void storeEvent(Event event) { + if (dropEventNames.contains(event.getClass().getName())) { + return; + } + try { + store.add(gson.toJson(event)); + } catch (IOException ex) { + log.error("Cannot add event to event store", ex); + } + } + + protected synchronized void sendAllPendingEvents() throws PermissionBackendException { + try { + long current = store.getHead(); + while (lastSent < current) { + long next = lastSent + 1; + fireEvent(gson.fromJson(store.get(next), Event.class)); + lastSent = next; + } + } catch (IOException e) { + // Next Event would re-try the events. + } + for (StreamEventListener l : streamEventListeners) { + l.onStreamEventUpdate(); + } + } + + protected void readAndParseCfg(String pluginName, GerritServerConfigProvider configProvider) { + PluginConfig cfg = PluginConfig.createFromGerritConfig(pluginName, configProvider.loadConfig()); + for (String filter : cfg.getStringList(KEY_FILTER)) { + String pieces[] = filter.split(" "); + if (pieces.length == 3 + && FILTER_TYPE_DROP.equals(pieces[0]) + && FILTER_ELEMENT_CLASSNAME.equals(pieces[1])) { + dropEventNames.add(pieces[2]); + } else { + log.error("Ignoring invalid filter: " + filter); + } + } + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/events/Module.java b/src/main/java/com/googlesource/gerrit/plugins/events/Module.java index 27830fb..c0e1271 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/events/Module.java +++ b/src/main/java/com/googlesource/gerrit/plugins/events/Module.java
@@ -14,8 +14,9 @@ package com.googlesource.gerrit.plugins.events; +import com.google.gerrit.extensions.registration.DynamicItem; import com.google.gerrit.extensions.registration.DynamicSet; -import com.google.gerrit.server.events.EventListener; +import com.google.gerrit.server.events.EventDispatcher; import com.google.inject.AbstractModule; import com.googlesource.gerrit.plugins.events.fsstore.FsStore; @@ -24,6 +25,6 @@ protected void configure() { DynamicSet.setOf(binder(), StreamEventListener.class); bind(EventStore.class).to(FsStore.class); - DynamicSet.bind(binder(), EventListener.class).to(CoreListener.class); + DynamicItem.bind(binder(), EventDispatcher.class).to(FileSystemEventBroker.class); } }