Support plugin Richdata APIs via implementation (#1581)

* Rich Data for Events Plugin

* plugin-api -> api

* use BinaryData as a root and add getValue to UInt256Value

* bring rick data changes in line with proposed APIs.

Undo orthagonal naming changes.

* add size

* use log and transaction interfaces from rich data api

* update to released plugin api version and add new event listener.

* add tests

* fix acceptance tests

Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>

acceptance-tests/src/test/java/tech/pegasys/pantheon/tests/acceptance/dsl/node/ThreadPantheonNodeRunner.java

@@ -64,7 +64,7 @@ public class ThreadPantheonNodeRunner implements PantheonNodeRunner {
   private final Map<Node, PantheonPluginContextImpl> pantheonPluginContextMap = new HashMap<>();
 
   private PantheonPluginContextImpl buildPluginContext(final PantheonNode node) {
-    PantheonPluginContextImpl pantheonPluginContext = new PantheonPluginContextImpl();
+    final PantheonPluginContextImpl pantheonPluginContext = new PantheonPluginContextImpl();
     final Path pluginsPath = node.homeDirectory().resolve("plugins");
     final File pluginsDirFile = pluginsPath.toFile();
     if (!pluginsDirFile.isDirectory()) {
@@ -126,7 +126,7 @@ public class ThreadPantheonNodeRunner implements PantheonNodeRunner {
 
     final RunnerBuilder runnerBuilder = new RunnerBuilder();
     if (node.getPermissioningConfiguration().isPresent()) {
-      PermissioningConfiguration permissioningConfiguration =
+      final PermissioningConfiguration permissioningConfiguration =
           node.getPermissioningConfiguration().get();
 
       runnerBuilder.permissioningConfiguration(permissioningConfiguration);
@@ -134,7 +134,9 @@ public class ThreadPantheonNodeRunner implements PantheonNodeRunner {
 
     pantheonPluginContext.addService(
         PantheonEvents.class,
-        new PantheonEventsImpl(pantheonController.getProtocolManager().getBlockBroadcaster()));
+        new PantheonEventsImpl(
+            pantheonController.getProtocolManager().getBlockBroadcaster(),
+            pantheonController.getTransactionPool()));
     pantheonPluginContext.startPlugins();
 
     final Runner runner =
@@ -167,7 +169,7 @@ public class ThreadPantheonNodeRunner implements PantheonNodeRunner {
 
   @Override
   public void stopNode(final PantheonNode node) {
-    PantheonPluginContextImpl pluginContext = pantheonPluginContextMap.remove(node);
+    final PantheonPluginContextImpl pluginContext = pantheonPluginContextMap.remove(node);
     if (pluginContext != null) {
       pluginContext.stopPlugins();
     }

consensus/common/build.gradle

@@ -26,6 +26,8 @@ jar {
 }
 
 dependencies {
+  api 'tech.pegasys.pantheon:plugin-api'
+
   implementation project(':ethereum:core')
   implementation project(':ethereum:jsonrpc')
   implementation project(':util')

crypto/build.gradle

@@ -29,6 +29,7 @@ dependencies {
   api project(':util')
 
   api 'org.bouncycastle:bcprov-jdk15on'
+  api 'tech.pegasys.pantheon:plugin-api'
 
   implementation 'com.google.guava:guava'
   implementation 'org.apache.logging.log4j:log4j-api'

ethereum/core/build.gradle

@@ -38,6 +38,7 @@ dependencies {
   implementation 'com.google.guava:guava'
   implementation 'io.vertx:vertx-core'
   implementation 'org.apache.logging.log4j:log4j-api'
+  implementation 'tech.pegasys.pantheon:plugin-api'
 
   runtime 'org.apache.logging.log4j:log4j-core'
 

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/Address.java

@@ -24,7 +24,8 @@ import tech.pegasys.pantheon.util.bytes.DelegatingBytesValue;
 import com.fasterxml.jackson.annotation.JsonCreator;
 
 /** A 160-bits account address. */
-public class Address extends DelegatingBytesValue {
+public class Address extends DelegatingBytesValue
+    implements tech.pegasys.pantheon.plugin.data.Address {
 
   public static final int SIZE = 20;
 

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/BlockHeader.java

@@ -23,7 +23,8 @@ import java.util.function.Supplier;
 import com.google.common.base.Suppliers;
 
 /** A mined Ethereum block header. */
-public class BlockHeader extends SealableBlockHeader {
+public class BlockHeader extends SealableBlockHeader
+    implements tech.pegasys.pantheon.plugin.data.BlockHeader {
 
   public static final int MAX_EXTRA_DATA_BYTES = 32;
 
@@ -79,6 +80,7 @@ public class BlockHeader extends SealableBlockHeader {
    *
    * @return the block mixed hash
    */
+  @Override
   public Hash getMixHash() {
     return mixHash;
   }
@@ -88,6 +90,7 @@ public class BlockHeader extends SealableBlockHeader {
    *
    * @return the block nonce
    */
+  @Override
   public long getNonce() {
     return nonce;
   }
@@ -109,6 +112,11 @@ public class BlockHeader extends SealableBlockHeader {
     return hash.get();
   }
 
+  @Override
+  public tech.pegasys.pantheon.plugin.data.Hash getBlockHash() {
+    return hash.get();
+  }
+
   /**
    * Write an RLP representation.
    *

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/Hash.java

@@ -22,7 +22,7 @@ import tech.pegasys.pantheon.util.bytes.DelegatingBytes32;
 import com.fasterxml.jackson.annotation.JsonCreator;
 
 /** A 32-bytes hash value as used in Ethereum blocks, that is the result of the KEC algorithm. */
-public class Hash extends DelegatingBytes32 {
+public class Hash extends DelegatingBytes32 implements tech.pegasys.pantheon.plugin.data.Hash {
 
   public static final Hash ZERO = new Hash(Bytes32.ZERO);
 
@@ -65,4 +65,14 @@ public class Hash extends DelegatingBytes32 {
   public static Hash fromHexStringLenient(final String str) {
     return new Hash(Bytes32.fromHexStringLenient(str));
   }
+
+  @Override
+  public byte[] getByteArray() {
+    return super.getByteArray();
+  }
+
+  @Override
+  public String getHexString() {
+    return super.getHexString();
+  }
 }

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/Log.java

@@ -26,7 +26,7 @@ import com.google.common.collect.ImmutableList;
  * A log entry is a tuple of a logger’s address (the address of the contract that added the logs), a
  * series of 32-bytes log topics, and some number of bytes of data.
  */
-public class Log {
+public class Log implements tech.pegasys.pantheon.plugin.data.Log {
 
   private final Address logger;
   private final BytesValue data;
@@ -71,14 +71,17 @@ public class Log {
     return new Log(logger, data, topics);
   }
 
+  @Override
   public Address getLogger() {
     return logger;
   }
 
+  @Override
   public BytesValue getData() {
     return data;
   }
 
+  @Override
   public List<LogTopic> getTopics() {
     return topics;
   }

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/LogsBloomFilter.java

@@ -17,6 +17,7 @@ import static tech.pegasys.pantheon.crypto.Hash.keccak256;
 
 import tech.pegasys.pantheon.ethereum.rlp.RLPException;
 import tech.pegasys.pantheon.ethereum.rlp.RLPInput;
+import tech.pegasys.pantheon.plugin.data.UnformattedData;
 import tech.pegasys.pantheon.util.bytes.BytesValue;
 import tech.pegasys.pantheon.util.bytes.MutableBytesValue;
 
@@ -31,7 +32,7 @@ import java.util.Collection;
  * corresponding double-bytes are: bd2b, 01af, cd27, corresponding to the following bits in the
  * bloom filter: 1323, 431, 1319
  */
-public class LogsBloomFilter {
+public class LogsBloomFilter implements UnformattedData {
 
   public static final int BYTE_SIZE = 256;
   private static final int LEAST_SIGNIFICANT_BYTE = 0xFF;
@@ -149,4 +150,19 @@ public class LogsBloomFilter {
   public String toString() {
     return data.toString();
   }
+
+  @Override
+  public byte[] getByteArray() {
+    return data.getByteArray();
+  }
+
+  @Override
+  public int size() {
+    return data.size();
+  }
+
+  @Override
+  public String getHexString() {
+    return data.getHexString();
+  }
 }

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/core/Transaction.java

@@ -30,7 +30,7 @@ import java.util.Objects;
 import java.util.Optional;
 
 /** An operation submitted by an external actor to be applied to the system. */
-public class Transaction {
+public class Transaction implements tech.pegasys.pantheon.plugin.data.Transaction {
 
   // Used for transactions that are not tied to a specific chain
   // (e.g. does not have a chain id associated with it).
@@ -154,6 +154,7 @@ public class Transaction {
    *
    * @return the transaction nonce
    */
+  @Override
   public long getNonce() {
     return nonce;
   }
@@ -163,6 +164,7 @@ public class Transaction {
    *
    * @return the transaction gas price
    */
+  @Override
   public Wei getGasPrice() {
     return gasPrice;
   }
@@ -172,6 +174,7 @@ public class Transaction {
    *
    * @return the transaction gas limit
    */
+  @Override
   public long getGasLimit() {
     return gasLimit;
   }
@@ -184,6 +187,7 @@ public class Transaction {
    *
    * @return the transaction recipient if a message call; otherwise {@code Optional.empty()}
    */
+  @Override
   public Optional<Address> getTo() {
     return to;
   }
@@ -193,6 +197,7 @@ public class Transaction {
    *
    * @return the value transferred in the transaction
    */
+  @Override
   public Wei getValue() {
     return value;
   }
@@ -215,6 +220,26 @@ public class Transaction {
     return payload;
   }
 
+  /**
+   * Returns the payload if this is a contract creation transaction.
+   *
+   * @return if present the init code
+   */
+  @Override
+  public Optional<BytesValue> getInit() {
+    return getTo().isPresent() ? Optional.empty() : Optional.of(payload);
+  }
+
+  /**
+   * Returns the payload if this is a message call transaction.
+   *
+   * @return if present the init code
+   */
+  @Override
+  public Optional<BytesValue> getData() {
+    return getTo().isPresent() ? Optional.of(payload) : Optional.empty();
+  }
+
   /**
    * Return the transaction chain id (if it exists)
    *
@@ -223,6 +248,7 @@ public class Transaction {
    *
    * @return the transaction chain id if it exists; otherwise {@code OptionalInt.empty()}
    */
+  @Override
   public Optional<BigInteger> getChainId() {
     return chainId;
   }
@@ -232,6 +258,7 @@ public class Transaction {
    *
    * @return the transaction sender
    */
+  @Override
   public Address getSender() {
     if (sender == null) {
       final SECP256K1.PublicKey publicKey =
@@ -279,18 +306,21 @@ public class Transaction {
     out.writeBigIntegerScalar(getSignature().getS());
   }
 
+  @Override
   public BigInteger getR() {
     return signature.getR();
   }
 
+  @Override
   public BigInteger getS() {
     return signature.getS();
   }
 
+  @Override
   public BigInteger getV() {
     final BigInteger v;
     final BigInteger recId = BigInteger.valueOf(signature.getRecId());
-    if (!chainId.isPresent()) {
+    if (chainId.isEmpty()) {
       v = recId.add(REPLAY_UNPROTECTED_V_BASE);
     } else {
       v = recId.add(REPLAY_PROTECTED_V_BASE).add(TWO.multiply(chainId.get()));
@@ -317,7 +347,7 @@ public class Transaction {
    * @return {@code true} if this is a contract-creation transaction; otherwise {@code false}
    */
   public boolean isContractCreation() {
-    return !getTo().isPresent();
+    return getTo().isEmpty();
   }
 
   /**
@@ -498,7 +528,7 @@ public class Transaction {
       return build();
     }
 
-    protected SECP256K1.Signature computeSignature(final SECP256K1.KeyPair keys) {
+    SECP256K1.Signature computeSignature(final SECP256K1.KeyPair keys) {
       final Bytes32 hash =
           computeSenderRecoveryHash(nonce, gasPrice, gasLimit, to, value, payload, chainId);
       return SECP256K1.sign(hash, keys);

ethereum/mock-p2p/build.gradle

@@ -26,6 +26,8 @@ jar {
 }
 
 dependencies {
+  api 'tech.pegasys.pantheon:plugin-api'
+
   implementation project(':ethereum:p2p')
   implementation project(':ethereum:permissioning')
   implementation project(':util')

ethereum/rlp/build.gradle

@@ -27,6 +27,7 @@ jar {
 
 dependencies {
   api project(':util')
+  api 'tech.pegasys.pantheon:plugin-api'
 
   implementation 'com.google.guava:guava'
   implementation 'io.vertx:vertx-core'

gradle/versions.gradle

@@ -89,6 +89,6 @@ dependencyManagement {
 
     dependency 'org.xerial.snappy:snappy-java:1.1.7.3'
 
-    dependency "tech.pegasys.pantheon:plugin-api:1.1.3"
+    dependency "tech.pegasys.pantheon:plugin-api:1.2.2"
   }
 }

pantheon/src/main/java/tech/pegasys/pantheon/cli/PantheonCommand.java

@@ -811,7 +811,9 @@ public class PantheonCommand implements DefaultCommandValues, Runnable {
   private PantheonCommand startPlugins() {
     pantheonPluginContext.addService(
         PantheonEvents.class,
-        new PantheonEventsImpl((pantheonController.getProtocolManager().getBlockBroadcaster())));
+        new PantheonEventsImpl(
+            (pantheonController.getProtocolManager().getBlockBroadcaster()),
+            pantheonController.getTransactionPool()));
     pantheonPluginContext.startPlugins();
     return this;
   }

pantheon/src/main/java/tech/pegasys/pantheon/services/PantheonEventsImpl.java

@@ -13,17 +13,19 @@
 package tech.pegasys.pantheon.services;
 
 import tech.pegasys.pantheon.ethereum.core.Block;
+import tech.pegasys.pantheon.ethereum.core.Transaction;
 import tech.pegasys.pantheon.ethereum.eth.sync.BlockBroadcaster;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPool;
 import tech.pegasys.pantheon.plugin.services.PantheonEvents;
 
-import com.google.common.collect.ImmutableMap;
-import io.vertx.core.json.Json;
-
 public class PantheonEventsImpl implements PantheonEvents {
   private final BlockBroadcaster blockBroadcaster;
+  private final TransactionPool transactionPool;
 
-  public PantheonEventsImpl(final BlockBroadcaster blockBroadcaster) {
+  public PantheonEventsImpl(
+      final BlockBroadcaster blockBroadcaster, final TransactionPool transactionPool) {
     this.blockBroadcaster = blockBroadcaster;
+    this.transactionPool = transactionPool;
   }
 
   @Override
@@ -41,13 +43,24 @@ public class PantheonEventsImpl implements PantheonEvents {
 
   private void dispatchNewBlockPropagatedMessage(
       final Block block, final NewBlockPropagatedListener listener) {
-    final ImmutableMap<Object, Object> result =
-        new ImmutableMap.Builder<>()
-            .put("type", "NewBlock")
-            .put("blockHash", block.getHash().toString())
-            .put("blockNumber", block.getHeader().getNumber())
-            .put("timestamp", block.getHeader().getTimestamp())
-            .build();
-    listener.newBlockPropagated(Json.encode(result));
+    listener.newBlockPropagated(block.getHeader());
+  }
+
+  @Override
+  public Object addNewTransactionAddedListener(final NewTransactionAddedListener listener) {
+    return transactionPool.subscribePendingTransactions(
+        transaction -> dispatchTransactionAddedMessage(transaction, listener));
+  }
+
+  @Override
+  public void removeNewTransactionAddedListener(final Object listenerIdentifier) {
+    if (listenerIdentifier instanceof Long) {
+      transactionPool.unsubscribePendingTransactions((Long) listenerIdentifier);
+    }
+  }
+
+  private void dispatchTransactionAddedMessage(
+      final Transaction transaction, final NewTransactionAddedListener listener) {
+    listener.newTransactionAdded(transaction);
   }
 }

pantheon/src/test/java/tech/pegasys/pantheon/plugins/TestPantheonEventsPlugin.java

@@ -14,6 +14,7 @@ package tech.pegasys.pantheon.plugins;
 
 import tech.pegasys.pantheon.plugin.PantheonContext;
 import tech.pegasys.pantheon.plugin.PantheonPlugin;
+import tech.pegasys.pantheon.plugin.data.BlockHeader;
 import tech.pegasys.pantheon.plugin.services.PantheonEvents;
 
 import java.io.File;
@@ -63,16 +64,16 @@ public class TestPantheonEventsPlugin implements PantheonPlugin {
     LOG.info("No longer listening with ID#" + subscriptionId);
   }
 
-  private void onBlockAnnounce(final String json) {
+  private void onBlockAnnounce(final BlockHeader header) {
     final int blockCount = blockCounter.incrementAndGet();
-    LOG.info("I got a new block! (I've seen {}) - {}", blockCount, json);
+    LOG.info("I got a new block! (I've seen {}) - {}", blockCount, header);
     try {
       final File callbackFile = new File(callbackDir, "newBlock." + blockCount);
       if (!callbackFile.getParentFile().exists()) {
         callbackFile.getParentFile().mkdirs();
         callbackFile.getParentFile().deleteOnExit();
       }
-      Files.write(callbackFile.toPath(), Collections.singletonList(json));
+      Files.write(callbackFile.toPath(), Collections.singletonList(header.toString()));
       callbackFile.deleteOnExit();
     } catch (final IOException ioe) {
       throw new RuntimeException(ioe);

pantheon/src/test/java/tech/pegasys/pantheon/services/PantheonEventsImplTest.java

@@ -13,17 +13,41 @@
 package tech.pegasys.pantheon.services;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.when;
 
+import tech.pegasys.pantheon.crypto.SECP256K1.KeyPair;
+import tech.pegasys.pantheon.ethereum.ProtocolContext;
+import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
 import tech.pegasys.pantheon.ethereum.core.Block;
 import tech.pegasys.pantheon.ethereum.core.BlockBody;
 import tech.pegasys.pantheon.ethereum.core.BlockHeaderTestFixture;
+import tech.pegasys.pantheon.ethereum.core.TransactionTestFixture;
+import tech.pegasys.pantheon.ethereum.core.Wei;
+import tech.pegasys.pantheon.ethereum.core.WorldState;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthMessages;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers;
+import tech.pegasys.pantheon.ethereum.eth.manager.EthScheduler;
 import tech.pegasys.pantheon.ethereum.eth.sync.BlockBroadcaster;
+import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPool;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPoolConfiguration;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPoolFactory;
+import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
+import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
+import tech.pegasys.pantheon.ethereum.mainnet.TransactionValidator;
+import tech.pegasys.pantheon.ethereum.mainnet.ValidationResult;
+import tech.pegasys.pantheon.ethereum.worldstate.WorldStateArchive;
+import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
+import tech.pegasys.pantheon.plugin.data.BlockHeader;
+import tech.pegasys.pantheon.plugin.data.Transaction;
+import tech.pegasys.pantheon.testutil.TestClock;
 import tech.pegasys.pantheon.util.uint.UInt256;
 
 import java.util.Collections;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
@@ -36,34 +60,76 @@ import org.mockito.junit.MockitoJUnitRunner;
 @RunWith(MockitoJUnitRunner.class)
 public class PantheonEventsImplTest {
 
-  @Mock private EthPeers ethPeers;
+  private static final KeyPair KEY_PAIR1 = KeyPair.generate();
+  private static final tech.pegasys.pantheon.ethereum.core.Transaction TX1 = createTransaction(1);
+  private static final tech.pegasys.pantheon.ethereum.core.Transaction TX2 = createTransaction(2);
+
+  @Mock private ProtocolSchedule<Void> mockProtocolSchedule;
+  @Mock private ProtocolContext<Void> mockProtocolContext;
+  @Mock private SyncState mockSyncState;
+  @Mock private EthPeers mockEthPeers;
   @Mock private EthContext mockEthContext;
+  @Mock private EthMessages mockEthMessages;
+  @Mock private EthScheduler mockEthScheduler;
+  @Mock private MutableBlockchain mockBlockchain;
+  @Mock private TransactionValidator mockTransactionValidator;
+  @Mock private ProtocolSpec<Void> mockProtocolSpec;
+  @Mock private WorldStateArchive mockWorldStateArchive;
+  @Mock private WorldState mockWorldState;
+  private tech.pegasys.pantheon.ethereum.core.BlockHeader fakeBlockHeader;
+  private TransactionPool transactionPool;
   private BlockBroadcaster blockBroadcaster;
   private PantheonEventsImpl serviceImpl;
 
   @Before
   public void setUp() {
-    when(ethPeers.streamAvailablePeers()).thenReturn(Stream.empty()).thenReturn(Stream.empty());
-    when(mockEthContext.getEthPeers()).thenReturn(ethPeers);
+    fakeBlockHeader =
+        new tech.pegasys.pantheon.ethereum.core.BlockHeader(
+            null, null, null, null, null, null, null, null, 1, 1, 1, 1, null, null, 1, null);
+
+    when(mockBlockchain.getBlockHeader(any())).thenReturn(Optional.of(fakeBlockHeader));
+    when(mockEthContext.getEthMessages()).thenReturn(mockEthMessages);
+    when(mockEthContext.getEthPeers()).thenReturn(mockEthPeers);
+    when(mockEthContext.getScheduler()).thenReturn(mockEthScheduler);
+    when(mockEthPeers.streamAvailablePeers()).thenReturn(Stream.empty()).thenReturn(Stream.empty());
+    when(mockProtocolContext.getBlockchain()).thenReturn(mockBlockchain);
+    when(mockProtocolContext.getWorldStateArchive()).thenReturn(mockWorldStateArchive);
+    when(mockProtocolSchedule.getByBlockNumber(anyLong())).thenReturn(mockProtocolSpec);
+    when(mockProtocolSpec.getTransactionValidator()).thenReturn(mockTransactionValidator);
+    when(mockTransactionValidator.validate(any())).thenReturn(ValidationResult.valid());
+    when(mockTransactionValidator.validateForSender(any(), any(), any()))
+        .thenReturn(ValidationResult.valid());
+    when(mockWorldStateArchive.get(any())).thenReturn(Optional.of(mockWorldState));
 
     blockBroadcaster = new BlockBroadcaster(mockEthContext);
-    serviceImpl = new PantheonEventsImpl(blockBroadcaster);
+    transactionPool =
+        TransactionPoolFactory.createTransactionPool(
+            mockProtocolSchedule,
+            mockProtocolContext,
+            mockEthContext,
+            TestClock.fixed(),
+            new NoOpMetricsSystem(),
+            mockSyncState,
+            Wei.ZERO,
+            TransactionPoolConfiguration.builder().build());
+
+    serviceImpl = new PantheonEventsImpl(blockBroadcaster, transactionPool);
   }
 
   @Test
-  public void eventFiresAfterSubscribe() {
-    final AtomicReference<String> result = new AtomicReference<>();
+  public void newBlockEventFiresAfterSubscribe() {
+    final AtomicReference<BlockHeader> result = new AtomicReference<>();
     serviceImpl.addNewBlockPropagatedListener(result::set);
 
     assertThat(result.get()).isNull();
     blockBroadcaster.propagate(generateBlock(), UInt256.of(1));
 
-    assertThat(result.get()).isNotEmpty();
+    assertThat(result.get()).isNotNull();
   }
 
   @Test
-  public void eventDoesNotFireAfterUnsubscribe() {
-    final AtomicReference<String> result = new AtomicReference<>();
+  public void newBlockEventDoesNotFireAfterUnsubscribe() {
+    final AtomicReference<BlockHeader> result = new AtomicReference<>();
     final Object id = serviceImpl.addNewBlockPropagatedListener(result::set);
 
     assertThat(result.get()).isNull();
@@ -82,14 +148,55 @@ public class PantheonEventsImplTest {
   }
 
   @Test
-  public void uselessUnsubscribesCompletes() {
+  public void newBlockEventUselessUnsubscribesCompletes() {
     serviceImpl.removeNewBlockPropagatedListener("doesNotExist");
     serviceImpl.removeNewBlockPropagatedListener(5);
     serviceImpl.removeNewBlockPropagatedListener(5L);
   }
 
+  @Test
+  public void newTransactionEventFiresAfterSubscribe() {
+    final AtomicReference<Transaction> result = new AtomicReference<>();
+    serviceImpl.addNewTransactionAddedListener(result::set);
+
+    assertThat(result.get()).isNull();
+    transactionPool.addLocalTransaction(TX1);
+
+    assertThat(result.get()).isNotNull();
+  }
+
+  @Test
+  public void newTransactionEventDoesNotFireAfterUnsubscribe() {
+    final AtomicReference<Transaction> result = new AtomicReference<>();
+    final Object id = serviceImpl.addNewTransactionAddedListener(result::set);
+
+    assertThat(result.get()).isNull();
+    transactionPool.addLocalTransaction(TX1);
+
+    serviceImpl.removeNewTransactionAddedListener(id);
+    result.set(null);
+
+    transactionPool.addLocalTransaction(TX2);
+    assertThat(result.get()).isNull();
+  }
+
+  @Test
+  public void newTransactionEventUselessUnsubscribesCompletes() {
+    serviceImpl.removeNewTransactionAddedListener("doesNotExist");
+    serviceImpl.removeNewTransactionAddedListener(5);
+    serviceImpl.removeNewTransactionAddedListener(5L);
+  }
+
   private Block generateBlock() {
     final BlockBody body = new BlockBody(Collections.emptyList(), Collections.emptyList());
     return new Block(new BlockHeaderTestFixture().buildHeader(), body);
   }
+
+  private static tech.pegasys.pantheon.ethereum.core.Transaction createTransaction(
+      final int transactionNumber) {
+    return new TransactionTestFixture()
+        .nonce(transactionNumber)
+        .gasLimit(0)
+        .createTransaction(KEY_PAIR1);
+  }
 }

services/kvstore/build.gradle

@@ -28,6 +28,8 @@ jar {
 dependencies {
   api project(':util')
 
+  api 'tech.pegasys.pantheon:plugin-api'
+
   implementation project(':metrics:core')
   implementation project(':metrics:rocksdb')
   implementation project(':services:util')

services/tasks/build.gradle

@@ -28,6 +28,8 @@ jar {
 dependencies {
   api project(':util')
 
+  api 'tech.pegasys.pantheon:plugin-api'
+
   compileOnly 'org.openjdk.jmh:jmh-generator-annprocess'
 
   implementation project(':metrics:core')

util/build.gradle

@@ -29,6 +29,7 @@ dependencies {
   implementation 'com.google.guava:guava'
   implementation 'io.vertx:vertx-core'
   implementation 'org.apache.logging.log4j:log4j-api'
+  implementation 'tech.pegasys.pantheon:plugin-api'
 
   runtime 'org.apache.logging.log4j:log4j-core'
 

util/src/main/java/tech/pegasys/pantheon/util/bytes/AbstractBytes32Backed.java

@@ -24,4 +24,9 @@ public class AbstractBytes32Backed implements Bytes32Backed {
   public Bytes32 getBytes() {
     return bytes;
   }
+
+  @Override
+  public int size() {
+    return bytes.size();
+  }
 }

util/src/main/java/tech/pegasys/pantheon/util/bytes/BytesBacked.java

@@ -12,8 +12,20 @@
  */
 package tech.pegasys.pantheon.util.bytes;
 
+import tech.pegasys.pantheon.plugin.data.BinaryData;
+
 /** Base interface for a value whose content is stored as bytes. */
-public interface BytesBacked {
+public interface BytesBacked extends BinaryData {
   /** @return The underlying backing bytes of the value. */
   BytesValue getBytes();
+
+  @Override
+  default byte[] getByteArray() {
+    return getBytes().getByteArray();
+  }
+
+  @Override
+  default String getHexString() {
+    return getBytes().getHexString();
+  }
 }

util/src/main/java/tech/pegasys/pantheon/util/bytes/BytesValue.java

@@ -15,6 +15,8 @@ package tech.pegasys.pantheon.util.bytes;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkElementIndex;
 
+import tech.pegasys.pantheon.plugin.data.UnformattedData;
+
 import java.nio.ByteBuffer;
 import java.security.MessageDigest;
 
@@ -33,7 +35,7 @@ import io.vertx.core.buffer.Buffer;
  *
  * @see BytesValues for static methods to create and work with {@link BytesValue}.
  */
-public interface BytesValue extends Comparable<BytesValue> {
+public interface BytesValue extends Comparable<BytesValue>, UnformattedData {
 
   /** The empty value (with 0 bytes). */
   BytesValue EMPTY = wrap(new byte[0]);
@@ -303,6 +305,7 @@ public interface BytesValue extends Comparable<BytesValue> {
   }
 
   /** @return The number of bytes this value represents. */
+  @Override
   int size();
 
   /**
@@ -514,6 +517,11 @@ public interface BytesValue extends Comparable<BytesValue> {
     return array;
   }
 
+  @Override
+  default byte[] getByteArray() {
+    return extractArray();
+  }
+
   /**
    * Get the bytes represented by this value as byte array.
    *
@@ -550,6 +558,16 @@ public interface BytesValue extends Comparable<BytesValue> {
   @Override
   String toString();
 
+  /**
+   * Returns the hexadecimal string representation of this value.
+   *
+   * @return The hexadecimal representation of this value, starting with "0x".
+   */
+  @Override
+  default String getHexString() {
+    return toString();
+  }
+
   default String toUnprefixedString() {
     final String prefixedHex = toString();
     return prefixedHex.startsWith("0x") ? prefixedHex.substring(2) : prefixedHex;

util/src/main/java/tech/pegasys/pantheon/util/uint/AbstractUInt256Value.java

@@ -18,6 +18,7 @@ import tech.pegasys.pantheon.util.bytes.AbstractBytes32Backed;
 import tech.pegasys.pantheon.util.bytes.Bytes32;
 import tech.pegasys.pantheon.util.bytes.Bytes32s;
 
+import java.math.BigInteger;
 import java.util.function.Supplier;
 
 /**
@@ -36,7 +37,7 @@ abstract class AbstractUInt256Value<T extends UInt256Value<T>> extends AbstractB
 
   private final Supplier<Counter<T>> mutableCtor;
 
-  protected AbstractUInt256Value(final Bytes32 bytes, final Supplier<Counter<T>> mutableCtor) {
+  AbstractUInt256Value(final Bytes32 bytes, final Supplier<Counter<T>> mutableCtor) {
     super(bytes);
     this.mutableCtor = mutableCtor;
   }
@@ -47,7 +48,7 @@ abstract class AbstractUInt256Value<T extends UInt256Value<T>> extends AbstractB
     return result.get();
   }
 
-  protected T binaryOp(final UInt256Value<?> value, final UInt256Bytes.BinaryOp op) {
+  T binaryOp(final UInt256Value<?> value, final UInt256Bytes.BinaryOp op) {
     final Counter<T> result = mutableCtor.get();
     op.applyOp(getBytes(), value.getBytes(), result.getBytes());
     return result.get();
@@ -193,4 +194,13 @@ abstract class AbstractUInt256Value<T extends UInt256Value<T>> extends AbstractB
   public String toString() {
     return UInt256Bytes.toString(getBytes());
   }
+
+  @Override
+  public Number getValue() {
+    if (UInt256Bytes.fitsLong(getBytes())) {
+      return toLong();
+    } else {
+      return new BigInteger(getByteArray());
+    }
+  }
 }

util/src/main/java/tech/pegasys/pantheon/util/uint/UInt256Value.java

@@ -14,6 +14,7 @@ package tech.pegasys.pantheon.util.uint;
 
 import static com.google.common.base.Preconditions.checkState;
 
+import tech.pegasys.pantheon.plugin.data.Quantity;
 import tech.pegasys.pantheon.util.bytes.Bytes32Backed;
 
 /**
@@ -41,7 +42,8 @@ import tech.pegasys.pantheon.util.bytes.Bytes32Backed;
  * @see BaseUInt256Value for a base class to extend in order to implement a {@link UInt256Value}.
  * @see Counter to obtain a mutable variant of a 256-bits integer.
  */
-public interface UInt256Value<T extends UInt256Value<T>> extends Bytes32Backed, Comparable<T> {
+public interface UInt256Value<T extends UInt256Value<T>>
+    extends Bytes32Backed, Comparable<T>, Quantity {
 
   int SIZE = 32;