Improve EthereumListener

ethereumj-core/src/main/java/org/ethereum/config/CommonConfig.java

@@ -17,6 +17,7 @@
  */
 package org.ethereum.config;
 
+import org.ethereum.core.EventDispatchThread;
 import org.ethereum.core.Repository;
 import org.ethereum.crypto.HashUtil;
 import org.ethereum.datasource.*;
@@ -27,6 +28,8 @@
 import org.ethereum.listener.CompositeEthereumListener;
 import org.ethereum.listener.EthereumListener;
 import org.ethereum.net.eth.handler.Eth63;
+import org.ethereum.publish.BackwardCompatibilityEthereumListenerProxy;
+import org.ethereum.publish.Publisher;
 import org.ethereum.sync.FastSyncManager;
 import org.ethereum.validator.*;
 import org.ethereum.vm.DataWord;
@@ -76,7 +79,8 @@ BeanPostProcessor initializer() {
     }
 
 
-    @Bean @Primary
+    @Bean
+    @Primary
     public Repository repository() {
         return new RepositoryWrapper();
     }
@@ -86,15 +90,16 @@ public Repository defaultRepository() {
         return new RepositoryRoot(stateSource(), null);
     }
 
-    @Bean @Scope("prototype")
+    @Bean
+    @Scope("prototype")
     public Repository repository(byte[] stateRoot) {
         return new RepositoryRoot(stateSource(), stateRoot);
     }
 
     /**
      * A source of nodes for state trie and all contract storage tries. <br/>
      * This source provides contract code too. <br/><br/>
-     *
+     * <p>
      * Picks node by 16-bytes prefix of its key. <br/>
      * Within {@link NodeKeyCompositor} this source is a part of ref counting workaround<br/><br/>
      *
@@ -126,7 +131,7 @@ public StateSource stateSource() {
     @Bean
     @Scope("prototype")
     public Source<byte[], byte[]> cachedDbSource(String name) {
-        AbstractCachedSource<byte[], byte[]>  writeCache = new AsyncWriteCache<byte[], byte[]>(blockchainSource(name)) {
+        AbstractCachedSource<byte[], byte[]> writeCache = new AsyncWriteCache<byte[], byte[]>(blockchainSource(name)) {
             @Override
             protected WriteCache<byte[], byte[]> createCache(Source<byte[], byte[]> source) {
                 WriteCache.BytesKey<byte[]> ret = new WriteCache.BytesKey<>(source, WriteCache.CacheType.SIMPLE);
@@ -166,7 +171,7 @@ public DbSource<byte[]> keyValueDataSource(String name, DbSettings settings) {
             DbSource<byte[]> dbSource;
             if ("inmem".equals(dataSource)) {
                 dbSource = new HashMapDB<>();
-            } else if ("leveldb".equals(dataSource)){
+            } else if ("leveldb".equals(dataSource)) {
                 dbSource = levelDbDataSource();
             } else {
                 dataSource = "rocksdb";
@@ -220,9 +225,20 @@ private void resetDataSource(Source source) {
         }
     }
 
-    @Bean(name = "EthereumListener")
-    public CompositeEthereumListener ethereumListener() {
-        return new CompositeEthereumListener();
+    @Bean
+    public Publisher publisher(EventDispatchThread eventDispatchThread) {
+        return new Publisher(eventDispatchThread);
+    }
+
+    @Bean
+    public CompositeEthereumListener compositeEthereumListener(EventDispatchThread eventDispatchThread) {
+        return new CompositeEthereumListener(eventDispatchThread);
+    }
+
+    @Bean
+    @Primary
+    public EthereumListener ethereumListener(CompositeEthereumListener compositeListener, Publisher publisher) {
+        return new BackwardCompatibilityEthereumListenerProxy(compositeListener, publisher);
     }
 
     @Bean
@@ -260,16 +276,18 @@ public byte[] serialize(byte[] object) {
                         ret.add(new DataWord(1));
                         return ret.getLast20Bytes();
                     }
+
                     public byte[] deserialize(byte[] stream) {
                         throw new RuntimeException("Shouldn't be called");
                     }
                 }, new Serializer<ProgramPrecompile, byte[]>() {
-                    public byte[] serialize(ProgramPrecompile object) {
-                        return object == null ? null : object.serialize();
-                    }
-                    public ProgramPrecompile deserialize(byte[] stream) {
-                        return stream == null ? null : ProgramPrecompile.deserialize(stream);
-                    }
+            public byte[] serialize(ProgramPrecompile object) {
+                return object == null ? null : object.serialize();
+            }
+
+            public ProgramPrecompile deserialize(byte[] stream) {
+                return stream == null ? null : ProgramPrecompile.deserialize(stream);
+            }
         });
     }
 
@@ -288,14 +306,14 @@ public DbFlushManager dbFlushManager() {
     }
 
     @Bean
-    public BlockHeaderValidator headerValidator() {
+    public BlockHeaderValidator headerValidator(SystemProperties systemProperties, Publisher publisher) {
 
         List<BlockHeaderRule> rules = new ArrayList<>(asList(
                 new GasValueRule(),
-                new ExtraDataRule(systemProperties()),
-                EthashRule.createRegular(systemProperties(), ethereumListener()),
-                new GasLimitRule(systemProperties()),
-                new BlockHashRule(systemProperties())
+                new ExtraDataRule(systemProperties),
+                EthashRule.createRegular(systemProperties, publisher),
+                new GasLimitRule(systemProperties),
+                new BlockHashRule(systemProperties)
         ));
 
         return new BlockHeaderValidator(rules);

ethereumj-core/src/main/java/org/ethereum/core/BlockchainImpl.java

@@ -24,13 +24,15 @@
 import org.ethereum.crypto.HashUtil;
 import org.ethereum.datasource.inmem.HashMapDB;
 import org.ethereum.db.*;
-import org.ethereum.trie.Trie;
-import org.ethereum.trie.TrieImpl;
 import org.ethereum.listener.EthereumListener;
-import org.ethereum.listener.EthereumListenerAdapter;
 import org.ethereum.manager.AdminInfo;
 import org.ethereum.sync.SyncManager;
-import org.ethereum.util.*;
+import org.ethereum.trie.Trie;
+import org.ethereum.trie.TrieImpl;
+import org.ethereum.util.AdvancedDeviceUtils;
+import org.ethereum.util.ByteUtil;
+import org.ethereum.util.FastByteComparisons;
+import org.ethereum.util.RLP;
 import org.ethereum.validator.DependentBlockHeaderRule;
 import org.ethereum.validator.ParentBlockHeaderValidator;
 import org.ethereum.vm.program.invoke.ProgramInvokeFactory;
@@ -47,20 +49,11 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.math.BigInteger;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.ConcurrentModificationException;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Set;
-import java.util.Stack;
+import java.util.*;
 
 import static java.lang.Math.max;
 import static java.lang.Runtime.getRuntime;
+import static java.lang.String.format;
 import static java.math.BigInteger.ONE;
 import static java.math.BigInteger.ZERO;
 import static java.util.Collections.emptyList;
@@ -110,7 +103,8 @@ public class BlockchainImpl implements Blockchain, org.ethereum.facade.Blockchai
     private static final int MAGIC_REWARD_OFFSET = 8;
     public static final byte[] EMPTY_LIST_HASH = sha3(RLP.encodeList(new byte[0]));
 
-    @Autowired @Qualifier("defaultRepository")
+    @Autowired
+    @Qualifier("defaultRepository")
     private Repository repository;
 
     @Autowired
@@ -126,7 +120,7 @@ public class BlockchainImpl implements Blockchain, org.ethereum.facade.Blockchai
     private BigInteger totalDifficulty = ZERO;
 
     @Autowired
-    private EthereumListener listener;
+    private EthereumListener listenerProxy;
 
     @Autowired
     ProgramInvokeFactory programInvokeFactory;
@@ -176,7 +170,9 @@ public class BlockchainImpl implements Blockchain, org.ethereum.facade.Blockchai
 
     private Stack<State> stateStack = new Stack<>();
 
-    /** Tests only **/
+    /**
+     * Tests only
+     **/
     public BlockchainImpl() {
     }
 
@@ -187,11 +183,11 @@ public BlockchainImpl(final SystemProperties config) {
     }
 
     //todo: autowire over constructor
-    public BlockchainImpl(final BlockStore blockStore, final Repository repository) {
+    public BlockchainImpl(final BlockStore blockStore, final Repository repository, EthereumListener listener) {
         this.blockStore = blockStore;
         this.repository = repository;
         this.adminInfo = new AdminInfo();
-        this.listener = new EthereumListenerAdapter();
+        this.listenerProxy = listener;
         this.parentHeaderValidator = null;
         this.transactionStore = new TransactionStore(new HashMapDB());
         this.eventDispatchThread = EventDispatchThread.getDefault();
@@ -209,11 +205,6 @@ public BlockchainImpl withAdminInfo(AdminInfo adminInfo) {
         return this;
     }
 
-    public BlockchainImpl withEthereumListener(EthereumListener listener) {
-        this.listener = listener;
-        return this;
-    }
-
     public BlockchainImpl withSyncManager(SyncManager syncManager) {
         this.syncManager = syncManager;
         return this;
@@ -455,8 +446,8 @@ public synchronized ImportResult tryToConnect(final Block block) {
         }
 
         if (ret.isSuccessful()) {
-            listener.onBlock(summary, ret == IMPORTED_BEST);
-            listener.trace(String.format("Block chain size: [ %d ]", this.getSize()));
+            listenerProxy.onBlock(summary, ret == IMPORTED_BEST);
+            listenerProxy.trace(format("Block chain size: [ %d ]", this.getSize()));
 
             if (ret == IMPORTED_BEST) {
                 eventDispatchThread.invokeLater(() -> pendingState.processBest(block, summary.getReceipts()));
@@ -493,7 +484,7 @@ public synchronized Block createNewBlock(Block parent, List<Transaction> txs, Li
                 new byte[0],  // nonce   (to mine)
                 new byte[0],  // receiptsRoot - computed after running all transactions
                 calcTxTrie(txs),    // TransactionsRoot - computed after running all transactions
-                new byte[] {0}, // stateRoot - computed after running all transactions
+                new byte[]{0}, // stateRoot - computed after running all transactions
                 txs,
                 null);  // uncle list
 
@@ -825,7 +816,7 @@ public static Set<ByteArrayWrapper> getAncestors(BlockStore blockStore, Block te
         if (!isParentBlock) {
             it = blockStore.getBlockByHash(it.getParentHash());
         }
-        while(it != null && it.getNumber() >= limitNum) {
+        while (it != null && it.getNumber() >= limitNum) {
             ret.add(new ByteArrayWrapper(it.getHash()));
             it = blockStore.getBlockByHash(it.getParentHash());
         }
@@ -839,7 +830,7 @@ public Set<ByteArrayWrapper> getUsedUncles(BlockStore blockStore, Block testedBl
         if (!isParentBlock) {
             it = blockStore.getBlockByHash(it.getParentHash());
         }
-        while(it.getNumber() > limitNum) {
+        while (it.getNumber() > limitNum) {
             for (BlockHeader uncle : it.getUncleList()) {
                 ret.add(new ByteArrayWrapper(uncle.getHash()));
             }
@@ -852,8 +843,7 @@ private BlockSummary processBlock(Repository track, Block block) {
 
         if (!block.isGenesis() && !config.blockChainOnly()) {
             return applyBlock(track, block);
-        }
-        else {
+        } else {
             return new BlockSummary(block, new HashMap<byte[], BigInteger>(), new ArrayList<TransactionReceipt>(), new ArrayList<TransactionExecutionSummary>());
         }
     }
@@ -876,7 +866,7 @@ private BlockSummary applyBlock(Repository track, Block block) {
 
             Repository txTrack = track.startTracking();
             TransactionExecutor executor = new TransactionExecutor(tx, block.getCoinbase(),
-                    txTrack, blockStore, programInvokeFactory, block, listener, totalGasUsed)
+                    txTrack, blockStore, programInvokeFactory, block, listenerProxy, totalGasUsed)
                     .withCommonConfig(commonConfig);
 
             executor.init();
@@ -953,7 +943,7 @@ private Map<byte[], BigInteger> addReward(Repository track, Block block, List<Tr
                         .multiply(BigInteger.valueOf(MAGIC_REWARD_OFFSET + uncle.getNumber() - block.getNumber()))
                         .divide(BigInteger.valueOf(MAGIC_REWARD_OFFSET));
 
-                track.addBalance(uncle.getCoinbase(),uncleReward);
+                track.addBalance(uncle.getCoinbase(), uncleReward);
                 BigInteger existingUncleReward = rewards.get(uncle.getCoinbase());
                 if (existingUncleReward == null) {
                     rewards.put(uncle.getCoinbase(), uncleReward);
@@ -1096,7 +1086,7 @@ private void recordBlock(Block block) {
 
     public void updateBlockTotDifficulties(long startFrom) {
         // no synchronization here not to lock instance for long period
-        while(true) {
+        while (true) {
             synchronized (this) {
                 ((IndexedBlockStore) blockStore).updateTotDifficulties(startFrom);
 
@@ -1120,7 +1110,7 @@ public void updateBlockTotDifficulties(long startFrom) {
                         }
                     }
 
-                    if (totalDifficulty.compareTo(maxTD) < 0)  {
+                    if (totalDifficulty.compareTo(maxTD) < 0) {
                         blockStore.reBranch(bestStoredBlock);
                         bestBlock = bestStoredBlock;
                         totalDifficulty = maxTD;
@@ -1216,8 +1206,9 @@ public Iterator<BlockHeader> getIteratorOfHeadersStartFrom(BlockIdentifier ident
     /**
      * Searches block in blockStore, if it's not found there
      * and headerStore is defined, searches blockHeader in it.
+     *
      * @param number block number
-     * @return  Block header
+     * @return Block header
      */
     private BlockHeader findHeaderByNumber(long number) {
         Block block = blockStore.getChainBlockByNumber(number);
@@ -1235,6 +1226,7 @@ private BlockHeader findHeaderByNumber(long number) {
     /**
      * Searches block in blockStore, if it's not found there
      * and headerStore is defined, searches blockHeader in it.
+     *
      * @param hash block hash
      * @return Block header
      */
@@ -1381,7 +1373,7 @@ public byte[] next() {
     }
 
     private class State {
-//        Repository savedRepo = repository;
+        //        Repository savedRepo = repository;
         byte[] root = repository.getRoot();
         Block savedBest = bestBlock;
         BigInteger savedTD = totalDifficulty;

ethereumj-core/src/main/java/org/ethereum/core/EventDispatchThread.java

@@ -35,7 +35,7 @@
  * Created by Anton Nashatyrev on 29.12.2015.
  */
 @Component
-public class EventDispatchThread {
+public class EventDispatchThread implements Executor {
     private static final Logger logger = LoggerFactory.getLogger("blockchain");
     private static EventDispatchThread eventDispatchThread;
 
@@ -67,6 +67,11 @@ public void invokeLater(Runnable r) {
         return eventDispatchThread;
     }
 
+    @Override
+    public void execute(Runnable command) {
+        invokeLater(command);
+    }
+
     public void invokeLater(final Runnable r) {
         if (executor.isShutdown()) return;
         if (counter++ % 1000 == 0) logStatus();