belliottsmith commented on code in PR #113: URL: https://github.com/apache/cassandra-accord/pull/113#discussion_r1745978635
########## accord-core/src/main/java/accord/impl/DefaultLocalListeners.java: ########## @@ -0,0 +1,452 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 accord.impl; + +import java.util.Arrays; +import java.util.EnumMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.BiFunction; + +import accord.api.LocalListeners; +import accord.api.RemoteListeners; +import accord.local.Command; +import accord.local.CommandStore; +import accord.local.Commands; +import accord.local.Node; +import accord.local.PreLoadContext; +import accord.local.SafeCommand; +import accord.local.SafeCommandStore; +import accord.local.SaveStatus; +import accord.primitives.TxnId; +import accord.utils.AsymmetricComparator; +import accord.utils.Invariants; +import accord.utils.btree.BTree; +import accord.utils.btree.BTreeRemoval; + +// TODO (required): evict to disk +public class DefaultLocalListeners implements LocalListeners +{ + public static class Factory implements LocalListeners.Factory + { + final RemoteListeners remoteListeners; + final NotifySink notifySink; + + public Factory(Node node) + { + this(node, DefaultNotifySink.INSTANCE); + } + + public Factory(Node node, NotifySink notifySink) + { + this.remoteListeners = node.remoteListeners(); + this.notifySink = notifySink; + } + + @Override + public LocalListeners create(CommandStore store) + { + return new DefaultLocalListeners(remoteListeners, notifySink); + } + } + + public interface NotifySink + { + void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener); + boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, ComplexListener listener); + } + + public static class DefaultNotifySink implements NotifySink + { + public static final DefaultNotifySink INSTANCE = new DefaultNotifySink(); + + @Override + public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listenerId) + { + SafeCommand listener = safeStore.ifLoadedAndInitialised(listenerId); + if (listener != null) Commands.listenerUpdate(safeStore, listener, safeCommand); + else + { + //noinspection SillyAssignment,ConstantConditions + safeStore = safeStore; // prevent use in lambda + TxnId updatedId = safeCommand.txnId(); + PreLoadContext context = PreLoadContext.contextFor(listenerId, updatedId); + safeStore.commandStore() + .execute(context, safeStore0 -> notify(safeStore0, listenerId, updatedId)) + .begin(safeStore.agent()); + } + } + + private static void notify(SafeCommandStore safeStore, TxnId listenerId, TxnId updatedId) + { + Commands.listenerUpdate(safeStore, safeStore.unsafeGet(listenerId), safeStore.unsafeGet(updatedId)); + } + + @Override + public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, ComplexListener listener) + { + return listener.notify(safeStore, safeCommand); + } + } + + static class TxnListeners extends TxnId + { + final SaveStatus await; + TxnId[] listeners = NO_TXNIDS; + int count; + + TxnListeners(TxnId txnId, SaveStatus await) + { + super(txnId); + this.await = await; + } + + public int compareListeners(TxnListeners that) + { + int c = this.compareTo(that); + if (c == 0) c = this.await.compareTo(that.await); + return c; + } + + public static int compareBefore(TxnId txnId, TxnListeners that) + { + int c = txnId.compareTo(that); + if (c == 0) c = -1; + return c; + } + + public static int compare(TxnId txnId, SaveStatus await, TxnListeners that, int ifEqual) + { + int c = txnId.compareTo(that); + if (c == 0) c = await.compareTo(that.await); + if (c == 0) c = ifEqual; + return c; + } + + void notify(NotifySink notifySink, SafeCommandStore safeStore, SafeCommand safeCommand) + { + trim(); + for (int i = 0 ; i < count ; ++i) + { + TxnId listenerId = listeners[i]; + notifySink.notify(safeStore, safeCommand, listenerId); + } + } + + private int trim() + { + Arrays.sort(listeners, 0, count); + int removedCount = 0; + for (int i = 1 ; i < count ; ++i) + { + if (listeners[i - 1].compareTo(listeners[i]) == 0) ++removedCount; Review Comment: I think this is to remove duplicates - when our buffer is full we sort then filter the list. This is to avoid updating a sorted list with each listener addition while keeping the size of the collection minimal without logically permitting duplicate entries. We do the same prior to notification. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
