Transaction queue

src/main/java/com/limechain/network/Network.java

@@ -13,7 +13,7 @@
 import com.limechain.network.protocol.ping.Ping;
 import com.limechain.network.protocol.sync.SyncService;
 import com.limechain.network.protocol.sync.pb.SyncMessage.BlockResponse;
-import com.limechain.network.protocol.transactions.TransactionsService;
+import com.limechain.network.protocol.transaction.TransactionsService;
 import com.limechain.network.protocol.warp.WarpSyncService;
 import com.limechain.network.protocol.warp.dto.WarpSyncResponse;
 import com.limechain.storage.DBConstants;

src/main/java/com/limechain/network/protocol/transactions/Transactions.java → src/main/java/com/limechain/network/protocol/transaction/Transactions.java

@@ -1,4 +1,4 @@
-package com.limechain.network.protocol.transactions;
+package com.limechain.network.protocol.transaction;
 
 import com.limechain.network.StrictProtocolBinding;
 

src/main/java/com/limechain/network/protocol/transactions/TransactionsController.java → src/main/java/com/limechain/network/protocol/transaction/TransactionsController.java

@@ -1,4 +1,4 @@
-package com.limechain.network.protocol.transactions;
+package com.limechain.network.protocol.transaction;
 
 import io.libp2p.core.Stream;
 

src/main/java/com/limechain/network/protocol/transactions/TransactionsEngine.java → src/main/java/com/limechain/network/protocol/transaction/TransactionsEngine.java

@@ -1,9 +1,13 @@
-package com.limechain.network.protocol.transactions;
+package com.limechain.network.protocol.transaction;
 
 import com.limechain.network.ConnectionManager;
-import com.limechain.network.protocol.transactions.scale.TransactionsReader;
-import com.limechain.network.protocol.transactions.scale.TransactionsWriter;
+import com.limechain.network.protocol.transaction.scale.TransactionsReader;
+import com.limechain.network.protocol.transaction.scale.TransactionsWriter;
+import com.limechain.network.protocol.warp.dto.BlockHeader;
+import com.limechain.network.protocol.warp.dto.Extrinsics;
 import com.limechain.network.protocol.warp.exception.ScaleEncodingException;
+import com.limechain.runtime.Runtime;
+import com.limechain.storage.block.BlockState;
 import com.limechain.sync.warpsync.SyncedState;
 import io.emeraldpay.polkaj.scale.ScaleCodecReader;
 import io.emeraldpay.polkaj.scale.ScaleCodecWriter;
@@ -98,10 +102,22 @@ private void handleHandshake(PeerId peerId, Stream stream) {
 
  private void handleTransactionMessage(byte[] message, PeerId peerId) {
  ScaleCodecReader reader = new ScaleCodecReader(message);
- byte[][] transactions = reader.read(new TransactionsReader());
+ Extrinsics[] transactions = reader.read(new TransactionsReader());
  log.log(Level.INFO, "Received " + transactions.length + " transactions from Peer "
  + peerId);
- //TODO Add transactions to data
+
+ final BlockHeader header = BlockState.getInstance().bestBlockHeader();
+ if (header == null) {
+ log.log(Level.WARNING, "No best block header found");
+ return;
+ }
+
+ final Runtime runtime = BlockState.getInstance().getRuntime(header.getHash());
+ if (runtime == null) {
+ log.log(Level.WARNING, "No runtime found for block header " + header.getHash());
+ }
+ //TODO Validate transaction using runtime and then add to transaction pool
+ // (depends on StateStorage and TrieState)
  }
 
  /**
@@ -126,7 +142,9 @@ public void writeHandshakeToStream(Stream stream, PeerId peerId) {
  public void writeTransactionsMessage(Stream stream, PeerId peerId) {
  ByteArrayOutputStream buf = new ByteArrayOutputStream();
  try (ScaleCodecWriter writer = new ScaleCodecWriter(buf)) {
- writer.write(new TransactionsWriter(), new byte[][]{new byte[]{}, new byte[]{}});
+ writer.write(new TransactionsWriter(), new Extrinsics[]{
+ new Extrinsics(new byte[]{}), new Extrinsics(new byte[]{})
+ });
  } catch (IOException e) {
  throw new ScaleEncodingException(e);
  }

src/main/java/com/limechain/network/protocol/transactions/TransactionsProtocol.java → src/main/java/com/limechain/network/protocol/transaction/TransactionsProtocol.java

@@ -1,4 +1,4 @@
-package com.limechain.network.protocol.transactions;
+package com.limechain.network.protocol.transaction;
 
 import com.limechain.network.ConnectionManager;
 import com.limechain.network.encoding.Leb128LengthFrameDecoder;
@@ -93,7 +93,6 @@ public void onException(Throwable cause) {
  connectionManager.closeTransactionsStream(stream);
  if (cause != null) {
  log.log(Level.WARNING, "Transactions Exception: " + cause.getMessage());
- cause.printStackTrace();
  } else {
  log.log(Level.WARNING, "Transactions Exception with unknown cause");
  }

src/main/java/com/limechain/network/protocol/transactions/TransactionsService.java → src/main/java/com/limechain/network/protocol/transaction/TransactionsService.java

@@ -1,4 +1,4 @@
-package com.limechain.network.protocol.transactions;
+package com.limechain.network.protocol.transaction;
 
 import com.limechain.network.ConnectionManager;
 import com.limechain.network.protocol.NetworkService;

src/main/java/com/limechain/network/protocol/transaction/scale/TransactionsReader.java

@@ -0,0 +1,18 @@
+package com.limechain.network.protocol.transaction.scale;
+
+import com.limechain.network.protocol.warp.dto.Extrinsics;
+import io.emeraldpay.polkaj.scale.ScaleCodecReader;
+import io.emeraldpay.polkaj.scale.ScaleReader;
+
+public class TransactionsReader implements ScaleReader<Extrinsics[]> {
+
+ @Override
+ public Extrinsics[] read(ScaleCodecReader reader) {
+ int size = reader.readCompactInt();
+ Extrinsics[] transactions = new Extrinsics[size];
+ for (int i = 0; i < size; i++) {
+ transactions[i] = new Extrinsics(reader.readByteArray());
+ }
+ return transactions;
+ }
+}

src/main/java/com/limechain/network/protocol/transaction/scale/TransactionsWriter.java

@@ -0,0 +1,18 @@
+package com.limechain.network.protocol.transaction.scale;
+
+import com.limechain.network.protocol.warp.dto.Extrinsics;
+import io.emeraldpay.polkaj.scale.ScaleCodecWriter;
+import io.emeraldpay.polkaj.scale.ScaleWriter;
+
+import java.io.IOException;
+
+public class TransactionsWriter implements ScaleWriter<Extrinsics[]> {
+
+ @Override
+ public void write(ScaleCodecWriter writer, Extrinsics[] transactions) throws IOException {
+ writer.writeCompact(transactions.length);
+ for (int i = 0; i < transactions.length; i++) {
+ writer.writeAsList(transactions[i].getExtrinsic());
+ }
+ }
+}

src/main/java/com/limechain/network/protocol/transaction/state/Pool.java

@@ -0,0 +1,37 @@
+package com.limechain.network.protocol.transaction.state;
+
+import com.limechain.utils.HashUtils;
+import lombok.NoArgsConstructor;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@NoArgsConstructor
+public class Pool {
+ final Map<byte[], ValidTransaction> transactions = new HashMap<>();
+
+ public ValidTransaction get(byte[] extrinisics) {
+ byte[] key = HashUtils.hashWithBlake2b(extrinisics);
+
+ return transactions.get(key);
+ }
+
+ public ValidTransaction[] transactions() {
+ return transactions.values().toArray(ValidTransaction[]::new);
+ }
+
+ public byte[] insert(ValidTransaction validTransaction){
+ byte[] key = HashUtils.hashWithBlake2b(validTransaction.getExtrinsic());
+ transactions.put(key, validTransaction);
+ return key;
+ }
+
+ public void removeExtrinsic(byte[] extrinsic){
+ byte[] key = HashUtils.hashWithBlake2b(extrinsic);
+ transactions.remove(key);
+ }
+
+ public int length(){
+ return transactions.size();
+ }
+}

src/main/java/com/limechain/network/protocol/transaction/state/TransactionState.java

@@ -0,0 +1,103 @@
+package com.limechain.network.protocol.transaction.state;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.extern.java.Log;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+
+@Log
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class TransactionState {
+ private static final TransactionState INSTANCE = new TransactionState();
+ private final Pool transactionPool = new Pool();
+ private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ @Getter
+ @Setter
+ private Queue<ValidTransaction> transactionQueue = new PriorityQueue<>();
+
+ public static TransactionState getInstance() {
+ return INSTANCE;
+ }
+
+ public void pushTransaction(ValidTransaction validTransaction) {
+ transactionQueue.add(validTransaction);
+ }
+
+ public ValidTransaction popTransaction() {
+ return transactionQueue.poll();
+ }
+
+ public ValidTransaction popTransactionWithTimer(long timeout) throws InterruptedException {
+ ValidTransaction validTransaction = popTransaction();
+ if (validTransaction != null) return validTransaction;
+
+ Future<ValidTransaction> futureTransaction = popFutureTransaction();
+
+ try {
+ return futureTransaction.get(timeout, java.util.concurrent.TimeUnit.MILLISECONDS);
+ } catch (ExecutionException e) {
+ log.severe("Error while waiting for transaction: " + e.getMessage());
+ } catch (TimeoutException e) {
+ futureTransaction.cancel(true);
+ } catch (InterruptedException e) {
+ if (Thread.interrupted())
+ throw new InterruptedException();
+ }
+
+ return null;
+ }
+
+ @NotNull
+ private Future<ValidTransaction> popFutureTransaction() {
+ return executor.submit(() -> {
+ ValidTransaction transaction = null;
+ while (transaction == null) {
+ transaction = popTransaction();
+ Thread.sleep(50);
+ }
+ return transaction;
+ });
+ }
+
+ public ValidTransaction peek() {
+ return transactionQueue.peek();
+ }
+
+ public ValidTransaction[] pending() {
+ return transactionQueue.toArray(new ValidTransaction[0]);
+ }
+
+ public ValidTransaction[] pendingInPool() {
+ return transactionPool.transactions();
+ }
+
+ public boolean exists(ValidTransaction validTransaction) {
+ return transactionQueue.contains(validTransaction);
+ }
+
+ public void removeExtrinsic(byte[] extrinsic) {
+ transactionPool.removeExtrinsic(extrinsic);
+ ValidTransaction transactionToBeRemoved = new ValidTransaction(extrinsic);
+ transactionQueue.remove(transactionToBeRemoved);
+ }
+
+ public void removeExtrinsicFromPool(byte[] extrinsic) {
+ transactionPool.removeExtrinsic(extrinsic);
+ }
+
+ public byte[] addToPool(ValidTransaction validTransaction) {
+ return transactionPool.insert(validTransaction);
+ }
+
+ //public void notifyStatus
+}

src/main/java/com/limechain/network/protocol/transaction/state/ValidTransaction.java

@@ -0,0 +1,37 @@
+package com.limechain.network.protocol.transaction.state;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.Comparator;
+
+@EqualsAndHashCode(onlyExplicitlyIncluded = true)
+public class ValidTransaction implements Comparable<ValidTransaction> {
+ @Getter
+ @EqualsAndHashCode.Include
+ private final byte[] extrinsic;
+ @Getter
+ private Validity validity;
+
+ public ValidTransaction(byte[] extrinsic){
+ this.extrinsic = extrinsic;
+ }
+
+ public ValidTransaction(byte[] extrinsic, Validity validity){
+ this.extrinsic = extrinsic;
+ this.validity = validity;
+ }
+
+ public int compareTo(@NotNull ValidTransaction transaction) {
+ return new ValidTransactionComparator().compare(this, transaction);
+ }
+
+ static class ValidTransactionComparator implements Comparator<ValidTransaction> {
+ public int compare(ValidTransaction validTransaction, ValidTransaction otherValidTransaction) {
+ return validTransaction.getValidity().getPriority()
+ .compareTo(otherValidTransaction.getValidity().getPriority());
+ }
+ }
+
+}

src/main/java/com/limechain/network/protocol/transaction/state/Validity.java

@@ -0,0 +1,24 @@
+package com.limechain.network.protocol.transaction.state;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.math.BigInteger;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class Validity {
+ private BigInteger priority;
+ private byte[][] requires;
+ private byte[][] provides;
+ private BigInteger longevity;
+ private boolean propagate;
+
+ public Validity(BigInteger priority){
+ this.priority = priority;
+ }
+}