revert

Author: thinkAfCod

hildr-node/src/main/java/io/optimism/l1/BeaconBlobFetcher.java

@@ -141,7 +141,7 @@ public List<BlobSidecar> getBlobSidecards(String blockId, final List<BigInteger>
                 : Map.of("indices", indices.stream().map(BigInteger::toString).collect(Collectors.joining(",")));
         var postfix = "%s%s".formatted(blockId, prepareQueryParams(params));
         var res = getBlobSidecars("%s/%s".formatted(this.sidecarsMethod, postfix));
-        if (res.getData() != null && !res.getData().isEmpty()) {
+        if (res != null && res.getData() != null && !res.getData().isEmpty()) {
             return res.getData();
         }
         if (this.archiverSidecarsMethod != null) {

hildr-node/src/main/java/io/optimism/l1/InnerWatcher.java

@@ -18,16 +18,12 @@
 import io.optimism.utilities.rpc.Web3jProvider;
 import io.optimism.utilities.telemetry.Logging;
 import io.optimism.utilities.telemetry.TracerTaskWrapper;
-import io.reactivex.BackpressureStrategy;
-import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import java.math.BigInteger;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.StructuredTaskScope;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -159,8 +155,6 @@ public class InnerWatcher extends AbstractExecutionThreadService {
 
     private boolean devnet = false;
 
-    private ScheduledThreadPoolExecutor scheduledExecutorService;
-
     /**
      * create a InnerWatcher instance.
      *
@@ -219,51 +213,23 @@ private void getMetadataFromL2(BigInteger l2StartBlock) {
         }
     }
 
-    private Disposable subscribeL1NewHeads() {
-        if (this.wsProvider != null) {
-            this.l1HeadListener = this.wsProvider
-                    .newHeadsNotifications()
-                    .subscribe(
-                            notification -> {
-                                NewHead header = notification.getParams().getResult();
-                                String hash = header.getHash();
-                                BigInteger number = Numeric.toBigInt(header.getNumber());
-                                String parentHash = header.getParentHash();
-                                BigInteger time = Numeric.toBigInt(header.getTimestamp());
-                                l1Head = new BlockInfo(hash, number, parentHash, time);
-                            },
-                            t -> {
-                                if (t instanceof WebsocketNotConnectedException) {
-                                    this.subscribeL1NewHeads();
-                                }
-                            });
-        } else {
-            this.scheduledExecutorService = new ScheduledThreadPoolExecutor(1);
-            this.l1HeadListener = Flowable.create(
-                            (subscriber) -> {
-                                this.scheduledExecutorService.scheduleAtFixedRate(
-                                        () -> {
-                                            EthBlock.Block block = null;
-                                            try {
-                                                block = pollBlock(
-                                                        this.provider, DefaultBlockParameterName.LATEST, false);
-                                            } catch (ExecutionException | InterruptedException e) {
-                                                LOGGER.warn("error while fetching L1 data for block", e);
-                                            }
-                                            subscriber.onNext(block);
-                                        },
-                                        0,
-                                        12,
-                                        TimeUnit.SECONDS);
-                            },
-                            BackpressureStrategy.BUFFER)
-                    .subscribe(notification -> {
-                        EthBlock.Block block = (EthBlock.Block) notification;
-                        l1Head = BlockInfo.from(block);
-                    });
-        }
-
-        return this.l1HeadListener;
+    private void subscribeL1NewHeads() {
+        this.l1HeadListener = this.wsProvider
+                .newHeadsNotifications()
+                .subscribe(
+                        notification -> {
+                            NewHead header = notification.getParams().getResult();
+                            String hash = header.getHash();
+                            BigInteger number = Numeric.toBigInt(header.getNumber());
+                            String parentHash = header.getParentHash();
+                            BigInteger time = Numeric.toBigInt(header.getTimestamp());
+                            l1Head = new BlockInfo(hash, number, parentHash, time);
+                        },
+                        t -> {
+                            if (t instanceof WebsocketNotConnectedException) {
+                                this.subscribeL1NewHeads();
+                            }
+                        });
     }
 
     /**
@@ -654,9 +620,6 @@ protected void shutDown() {
         if (this.wsProvider != null) {
             this.wsProvider.shutdown();
         }
-        if (this.scheduledExecutorService != null) {
-            this.scheduledExecutorService.shutdown();
-        }
     }
 
     @Override