tmp

Consensys · Oct 16, 2024 · 5bc80db · 5bc80db
1 parent f5790a5
commit 5bc80db
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 2 deletions.
diff --git a/...rking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/gossip/BlockGossipChannel.java b/...rking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/gossip/BlockGossipChannel.java
@@ -13,9 +13,11 @@
 
 package tech.pegasys.teku.networking.eth2.gossip;
 
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
+import tech.pegasys.teku.infrastructure.events.ChannelInterface;
 import tech.pegasys.teku.infrastructure.events.VoidReturningChannelInterface;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
 
-public interface BlockGossipChannel extends VoidReturningChannelInterface {
-  void publishBlock(SignedBeaconBlock block);
+public interface BlockGossipChannel extends ChannelInterface {
+  SafeFuture<Void> publishBlock(SignedBeaconBlock block);
 }
diff --git a/.../eth2/src/main/java/tech/pegasys/teku/networking/eth2/gossip/forks/GossipForkManager.java b/.../eth2/src/main/java/tech/pegasys/teku/networking/eth2/gossip/forks/GossipForkManager.java
@@ -26,9 +26,11 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.SpecMilestone;
@@ -246,6 +248,22 @@ private <T> void publishMessage(
                     slot));
   }
 
+  private <T> SafeFuture<Void> publishMessageWithFeedback(
+          final UInt64 slot,
+          final T message,
+          final String type,
+          final BiFunction<GossipForkSubscriptions, T, SafeFuture<Void>> publisher) {
+    final Optional<GossipForkSubscriptions> gossipForkSubscriptions = getSubscriptionActiveAtSlot(slot)
+            .filter(this::isActive);
+
+    if(gossipForkSubscriptions.isEmpty()) {
+      LOG.warn("Not publishing {} because no gossip subscriptions are active for slot {}", type, slot);
+      return SafeFuture.COMPLETE;
+    }
+
+    return publisher.apply(gossipForkSubscriptions.get(), message);
+  }
+
   public synchronized void subscribeToAttestationSubnetId(final int subnetId) {
     if (currentAttestationSubnets.add(subnetId)) {
       activeSubscriptions.forEach(