Add txpool_besuPendingTransactions API (#1196)

Add txpool_besuPendingTransactions API which add options to filter pending transactions.

Signed-off-by: Karim TAAM <karim.t2am@gmail.com>

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/RpcMethod.java

@@ -132,6 +132,7 @@ public enum RpcMethod {
   TRACE_TRANSACTION("trace_transaction"),
   TX_POOL_BESU_STATISTICS("txpool_besuStatistics"),
   TX_POOL_BESU_TRANSACTIONS("txpool_besuTransactions"),
+  TX_POOL_BESU_PENDING_TRANSACTIONS("txpool_besuPendingTransactions"),
   WEB3_CLIENT_VERSION("web3_clientVersion"),
   WEB3_SHA3("web3_sha3"),
   PLUGINS_RELOAD_CONFIG("plugins_reloadPluginConfig");

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/TxPoolBesuPendingTransactions.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.RpcMethod;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.PendingTransactionsParams;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.TransactionPendingResult;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.Filter;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class TxPoolBesuPendingTransactions implements JsonRpcMethod {
+
+  final PendingTransactionFilter pendingTransactionFilter;
+
+  private final PendingTransactions pendingTransactions;
+
+  public TxPoolBesuPendingTransactions(final PendingTransactions pendingTransactions) {
+    this.pendingTransactions = pendingTransactions;
+    this.pendingTransactionFilter = new PendingTransactionFilter();
+  }
+
+  @Override
+  public String getName() {
+    return RpcMethod.TX_POOL_BESU_PENDING_TRANSACTIONS.getMethodName();
+  }
+
+  @Override
+  public JsonRpcResponse response(final JsonRpcRequestContext requestContext) {
+
+    final Integer limit = requestContext.getRequiredParameter(0, Integer.class);
+    final List<Filter> filters =
+        requestContext
+            .getOptionalParameter(1, PendingTransactionsParams.class)
+            .map(PendingTransactionsParams::filters)
+            .orElse(Collections.emptyList());
+
+    final Set<Transaction> pendingTransactionsFiltered =
+        pendingTransactionFilter.reduce(pendingTransactions.getTransactionInfo(), filters, limit);
+
+    return new JsonRpcSuccessResponse(
+        requestContext.getRequest().getId(),
+        pendingTransactionsFiltered.stream()
+            .map(TransactionPendingResult::new)
+            .collect(Collectors.toSet()));
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/parameters/PendingTransactionsParams.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters;
+
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.FROM_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.GAS_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.GAS_PRICE_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.NONCE_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.TO_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.VALUE_FIELD;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.ACTION;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.EQ;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.exception.InvalidJsonRpcParameters;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.Filter;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class PendingTransactionsParams {
+
+  private final Map<String, String> from, to, gas, gasPrice, value, nonce;
+
+  @JsonCreator()
+  public PendingTransactionsParams(
+      @JsonProperty(FROM_FIELD) final Map<String, String> from,
+      @JsonProperty(TO_FIELD) final Map<String, String> to,
+      @JsonProperty(GAS_FIELD) final Map<String, String> gas,
+      @JsonProperty(GAS_PRICE_FIELD) final Map<String, String> gasPrice,
+      @JsonProperty(VALUE_FIELD) final Map<String, String> value,
+      @JsonProperty(NONCE_FIELD) final Map<String, String> nonce) {
+    this.from = from;
+    this.to = to;
+    this.gas = gas;
+    this.gasPrice = gasPrice;
+    this.value = value;
+    this.nonce = nonce;
+  }
+
+  public List<Filter> filters() throws IllegalArgumentException {
+    final List<Filter> createdFilters = new ArrayList<>();
+    getFilter(FROM_FIELD, from).ifPresent(createdFilters::add);
+    getFilter(TO_FIELD, to).ifPresent(createdFilters::add);
+    getFilter(GAS_FIELD, gas).ifPresent(createdFilters::add);
+    getFilter(GAS_PRICE_FIELD, gasPrice).ifPresent(createdFilters::add);
+    getFilter(VALUE_FIELD, value).ifPresent(createdFilters::add);
+    getFilter(NONCE_FIELD, nonce).ifPresent(createdFilters::add);
+    return createdFilters;
+  }
+
+  /**
+   * This method allows to retrieve a list of filters related to a key
+   *
+   * @param key the key that will be linked to the filters
+   * @param map the list of filters to parse
+   * @return the list of filters
+   */
+  private Optional<Filter> getFilter(final String key, final Map<String, String> map) {
+    if (map != null) {
+      if (map.size() > 1) {
+        throw new InvalidJsonRpcParameters("Only one operator per filter type allowed");
+      } else if (!map.isEmpty()) {
+        final Map.Entry<String, String> foundEntry = map.entrySet().stream().findFirst().get();
+        final Predicate predicate =
+            Predicate.fromValue(foundEntry.getKey().toUpperCase())
+                .orElseThrow(
+                    () ->
+                        new InvalidJsonRpcParameters(
+                            "Unknown field expected one of `eq`, `gt`, `lt`, `action`"));
+
+        final Filter filter = new Filter(key, foundEntry.getValue(), predicate);
+        if (key.equals(FROM_FIELD) && !predicate.equals(EQ)) {
+          throw new InvalidJsonRpcParameters("The `from` filter only supports the `eq` operator");
+        } else if (key.equals(TO_FIELD) && !predicate.equals(EQ) && !predicate.equals(ACTION)) {
+          throw new InvalidJsonRpcParameters(
+              "The `to` filter only supports the `eq` or `action` operator");
+        } else if (!key.equals(TO_FIELD) && predicate.equals(ACTION)) {
+          throw new InvalidJsonRpcParameters(
+              "The operator `action` is only supported by the `to` filter");
+        }
+        return Optional.of(filter);
+      }
+    }
+    return Optional.empty();
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/PendingTransactionsResult.java

@@ -27,9 +27,7 @@ public class PendingTransactionsResult implements TransactionResult {
 
   public PendingTransactionsResult(final Set<TransactionInfo> transactionInfoSet) {
     transactionInfoResults =
-        transactionInfoSet.stream()
+        transactionInfoSet.stream().map(TransactionInfoResult::new).collect(Collectors.toSet());
-            .map(t -> new TransactionInfoResult(t))
-            .collect(Collectors.toSet());
   }
 
   @JsonValue

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/transaction/pool/PendingTransactionFilter.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool;
+
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.ACTION;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.EQ;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.exception.InvalidJsonRpcParameters;
+import org.hyperledger.besu.ethereum.core.Address;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.core.Wei;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions.TransactionInfo;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class allows to filter a list of pending transactions
+ *
+ * <p>Here is the list of fields that can be used to filter a transaction : from, to, gas, gasPrice,
+ * value, nonce
+ */
+public class PendingTransactionFilter {
+
+  public static final String FROM_FIELD = "from";
+  public static final String TO_FIELD = "to";
+  public static final String GAS_FIELD = "gas";
+  public static final String GAS_PRICE_FIELD = "gasPrice";
+  public static final String VALUE_FIELD = "value";
+  public static final String NONCE_FIELD = "nonce";
+
+  public Set<Transaction> reduce(
+      final Set<TransactionInfo> pendingTransactions, final List<Filter> filters, final int limit)
+      throws InvalidJsonRpcParameters {
+    return pendingTransactions.stream()
+        .filter(transactionInfo -> applyFilters(transactionInfo, filters))
+        .limit(limit)
+        .map(TransactionInfo::getTransaction)
+        .collect(Collectors.toSet());
+  }
+
+  private boolean applyFilters(final TransactionInfo transactionInfo, final List<Filter> filters)
+      throws InvalidJsonRpcParameters {
+    boolean isValid = true;
+    for (Filter filter : filters) {
+      final Predicate predicate = filter.getPredicate();
+      final String value = filter.getFieldValue();
+      switch (filter.getFieldName()) {
+        case FROM_FIELD:
+          isValid = validateFrom(transactionInfo, predicate, value);
+          break;
+        case TO_FIELD:
+          isValid = validateTo(transactionInfo, predicate, value);
+          break;
+        case GAS_PRICE_FIELD:
+          isValid = validateWei(transactionInfo.getTransaction().getGasPrice(), predicate, value);
+          break;
+        case GAS_FIELD:
+          isValid =
+              validateWei(Wei.of(transactionInfo.getTransaction().getGasLimit()), predicate, value);
+          break;
+        case VALUE_FIELD:
+          isValid = validateWei(transactionInfo.getTransaction().getValue(), predicate, value);
+          break;
+        case NONCE_FIELD:
+          isValid = validateNonce(transactionInfo, predicate, value);
+          break;
+      }
+      if (!isValid) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private boolean validateFrom(
+      final TransactionInfo transactionInfo, final Predicate predicate, final String value)
+      throws InvalidJsonRpcParameters {
+    return predicate
+        .getOperator()
+        .apply(transactionInfo.getTransaction().getSender(), Address.fromHexString(value));
+  }
+
+  private boolean validateTo(
+      final TransactionInfo transactionInfo, final Predicate predicate, final String value)
+      throws InvalidJsonRpcParameters {
+    final Optional<Address> maybeTo = transactionInfo.getTransaction().getTo();
+    if (maybeTo.isPresent() && predicate.equals(EQ)) {
+      return predicate.getOperator().apply(maybeTo.get(), Address.fromHexString(value));
+    } else if (predicate.equals(ACTION)) {
+      return transactionInfo.getTransaction().isContractCreation();
+    }
+    return false;
+  }
+
+  private boolean validateNonce(
+      final TransactionInfo transactionInfo, final Predicate predicate, final String value)
+      throws InvalidJsonRpcParameters {
+    return predicate
+        .getOperator()
+        .apply(transactionInfo.getTransaction().getNonce(), Long.decode(value));
+  }
+
+  private boolean validateWei(
+      final Wei transactionWei, final Predicate predicate, final String value)
+      throws InvalidJsonRpcParameters {
+    return predicate.getOperator().apply(transactionWei, Wei.fromHexString(value));
+  }
+
+  public static class Filter {
+
+    private final String fieldName;
+    private final String fieldValue;
+    private final Predicate predicate;
+
+    public Filter(final String fieldName, final String fieldValue, final Predicate predicate) {
+      this.fieldName = fieldName;
+      this.fieldValue = fieldValue;
+      this.predicate = predicate;
+    }
+
+    public String getFieldName() {
+      return fieldName;
+    }
+
+    public String getFieldValue() {
+      return fieldValue;
+    }
+
+    public Predicate getPredicate() {
+      return predicate;
+    }
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/transaction/pool/Predicate.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+/**
+ * This class describes the behavior of predicates that can be used to filter pending transactions
+ */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public enum Predicate {
+  EQ((left, right) -> left.compareTo(right) == 0),
+  LT((left, right) -> left.compareTo(right) < 0),
+  GT((left, right) -> left.compareTo(right) > 0),
+  ACTION((left, right) -> false);
+
+  private final Operator operator;
+
+  Predicate(final Operator predicate) {
+    this.operator = predicate;
+  }
+
+  public Operator getOperator() {
+    return operator;
+  }
+
+  @FunctionalInterface
+  public interface Operator {
+    boolean apply(final Comparable left, final Comparable right);
+  }
+
+  public static Optional<Predicate> fromValue(final String value) {
+    return Stream.of(values()).filter(predicate -> predicate.name().equals(value)).findFirst();
+  }
+}

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/methods/TxPoolJsonRpcMethods.java

@@ -17,6 +17,7 @@ package org.hyperledger.besu.ethereum.api.jsonrpc.methods;
 import org.hyperledger.besu.ethereum.api.jsonrpc.RpcApi;
 import org.hyperledger.besu.ethereum.api.jsonrpc.RpcApis;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.TxPoolBesuPendingTransactions;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.TxPoolBesuStatistics;
 import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.TxPoolBesuTransactions;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
@@ -40,6 +41,7 @@ public class TxPoolJsonRpcMethods extends ApiGroupJsonRpcMethods {
   protected Map<String, JsonRpcMethod> create() {
     return mapOf(
         new TxPoolBesuTransactions(transactionPool.getPendingTransactions()),
+        new TxPoolBesuPendingTransactions(transactionPool.getPendingTransactions()),
         new TxPoolBesuStatistics(transactionPool.getPendingTransactions()));
   }
 }

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/TxPoolBesuPendingTransactionsTest.java

@@ -0,0 +1,259 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequest;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.exception.InvalidJsonRpcParameters;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.PendingTransactionsParams;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.TransactionPendingResult;
+import org.hyperledger.besu.ethereum.core.Address;
+import org.hyperledger.besu.ethereum.core.Hash;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.core.Wei;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
+
+import java.math.BigInteger;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.tuweni.bytes.Bytes;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@SuppressWarnings("unchecked")
+@RunWith(MockitoJUnitRunner.class)
+public class TxPoolBesuPendingTransactionsTest {
+
+  @Mock private PendingTransactions pendingTransactions;
+  private TxPoolBesuPendingTransactions method;
+  private final String JSON_RPC_VERSION = "2.0";
+  private final String TXPOOL_PENDING_TRANSACTIONS_METHOD = "txpool_besuPendingTransactions";
+
+  @Before
+  public void setUp() {
+    final Set<PendingTransactions.TransactionInfo> listTrx = getPendingTransactions();
+    method = new TxPoolBesuPendingTransactions(pendingTransactions);
+    when(this.pendingTransactions.getTransactionInfo()).thenReturn(listTrx);
+  }
+
+  @Test
+  public void returnsCorrectMethodName() {
+    assertThat(method.getName()).isEqualTo(TXPOOL_PENDING_TRANSACTIONS_METHOD);
+  }
+
+  @Test
+  public void shouldReturnPendingTransactions() {
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION, TXPOOL_PENDING_TRANSACTIONS_METHOD, new Object[] {100}));
+
+    final JsonRpcSuccessResponse actualResponse = (JsonRpcSuccessResponse) method.response(request);
+    final Set<TransactionPendingResult> result =
+        (Set<TransactionPendingResult>) actualResponse.getResult();
+    assertThat(result.size()).isEqualTo(4);
+  }
+
+  @Test
+  public void shouldReturnPendingTransactionsWithLimit() {
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION, TXPOOL_PENDING_TRANSACTIONS_METHOD, new Object[] {1}));
+
+    final JsonRpcSuccessResponse actualResponse = (JsonRpcSuccessResponse) method.response(request);
+
+    final Set<TransactionPendingResult> result =
+        (Set<TransactionPendingResult>) actualResponse.getResult();
+    assertThat(result.size()).isEqualTo(1);
+  }
+
+  @Test
+  public void shouldReturnPendingTransactionsWithFilter() {
+
+    final Map<String, String> fromFilter = new HashMap<>();
+    fromFilter.put("eq", "0x0000000000000000000000000000000000000001");
+
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION,
+                TXPOOL_PENDING_TRANSACTIONS_METHOD,
+                new Object[] {
+                  100,
+                  new PendingTransactionsParams(
+                      fromFilter,
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>())
+                }));
+
+    final JsonRpcSuccessResponse actualResponse = (JsonRpcSuccessResponse) method.response(request);
+
+    final Set<TransactionPendingResult> result =
+        (Set<TransactionPendingResult>) actualResponse.getResult();
+    assertThat(result.size()).isEqualTo(1);
+  }
+
+  @Test
+  public void shouldReturnsErrorIfInvalidPredicate() {
+
+    final Map<String, String> fromFilter = new HashMap<>();
+    fromFilter.put("invalid", "0x0000000000000000000000000000000000000001");
+
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION,
+                TXPOOL_PENDING_TRANSACTIONS_METHOD,
+                new Object[] {
+                  100,
+                  new PendingTransactionsParams(
+                      fromFilter,
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>())
+                }));
+
+    assertThatThrownBy(() -> method.response(request))
+        .isInstanceOf(InvalidJsonRpcParameters.class)
+        .hasMessageContaining("Unknown field expected one of `eq`, `gt`, `lt`, `action`");
+  }
+
+  @Test
+  public void shouldReturnsErrorIfInvalidNumberOfPredicate() {
+
+    final Map<String, String> fromFilter = new HashMap<>();
+    fromFilter.put("eq", "0x0000000000000000000000000000000000000001");
+    fromFilter.put("lt", "0x0000000000000000000000000000000000000001");
+
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION,
+                TXPOOL_PENDING_TRANSACTIONS_METHOD,
+                new Object[] {
+                  100,
+                  new PendingTransactionsParams(
+                      fromFilter,
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>())
+                }));
+
+    assertThatThrownBy(() -> method.response(request))
+        .isInstanceOf(InvalidJsonRpcParameters.class)
+        .hasMessageContaining("Only one operator per filter type allowed");
+  }
+
+  @Test
+  public void shouldReturnsErrorIfInvalidPredicateUsedForFromField() {
+
+    final Map<String, String> fromFilter = new HashMap<>();
+    fromFilter.put("lt", "0x0000000000000000000000000000000000000001");
+
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION,
+                TXPOOL_PENDING_TRANSACTIONS_METHOD,
+                new Object[] {
+                  100,
+                  new PendingTransactionsParams(
+                      fromFilter,
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>())
+                }));
+
+    assertThatThrownBy(() -> method.response(request))
+        .isInstanceOf(InvalidJsonRpcParameters.class)
+        .hasMessageContaining("The `from` filter only supports the `eq` operator");
+  }
+
+  @Test
+  public void shouldReturnsErrorIfInvalidPredicateUsedForToField() {
+
+    final Map<String, String> toFilter = new HashMap<>();
+    toFilter.put("lt", "0x0000000000000000000000000000000000000001");
+
+    final JsonRpcRequestContext request =
+        new JsonRpcRequestContext(
+            new JsonRpcRequest(
+                JSON_RPC_VERSION,
+                TXPOOL_PENDING_TRANSACTIONS_METHOD,
+                new Object[] {
+                  100,
+                  new PendingTransactionsParams(
+                      new HashMap<>(),
+                      toFilter,
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>(),
+                      new HashMap<>())
+                }));
+
+    assertThatThrownBy(() -> method.response(request))
+        .isInstanceOf(InvalidJsonRpcParameters.class)
+        .hasMessageContaining("The `to` filter only supports the `eq` or `action` operator");
+  }
+
+  private Set<PendingTransactions.TransactionInfo> getPendingTransactions() {
+    final List<PendingTransactions.TransactionInfo> transactionInfoList = new ArrayList<>();
+    for (int i = 1; i < 5; i++) {
+      Transaction transaction = mock(Transaction.class);
+      when(transaction.getGasPrice()).thenReturn(Wei.of(i));
+      when(transaction.getValue()).thenReturn(Wei.of(i));
+      when(transaction.getGasLimit()).thenReturn((long) i);
+      when(transaction.getNonce()).thenReturn((long) i);
+      when(transaction.getPayload()).thenReturn(Bytes.EMPTY);
+      when(transaction.getV()).thenReturn(BigInteger.ONE);
+      when(transaction.getR()).thenReturn(BigInteger.ONE);
+      when(transaction.getS()).thenReturn(BigInteger.ONE);
+      when(transaction.getSender()).thenReturn(Address.fromHexString(String.valueOf(i)));
+      when(transaction.getTo())
+          .thenReturn(Optional.of(Address.fromHexString(String.valueOf(i + 1))));
+      when(transaction.getHash()).thenReturn(Hash.fromHexStringLenient(String.valueOf(i)));
+      transactionInfoList.add(
+          new PendingTransactions.TransactionInfo(
+              transaction, true, Instant.ofEpochSecond(Integer.MAX_VALUE)));
+    }
+    return new LinkedHashSet<>(transactionInfoList);
+  }
+}

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/transaction/pool/PendingTransactionFilterTest.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.EMPTY_LIST;
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.ACTION;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.EQ;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.GT;
+import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.Predicate.LT;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.transaction.pool.PendingTransactionFilter.Filter;
+import org.hyperledger.besu.ethereum.core.Address;
+import org.hyperledger.besu.ethereum.core.Hash;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.core.Wei;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class PendingTransactionFilterTest {
+
+  @Parameterized.Parameters
+  public static Collection<Object[]> data() {
+    return asList(
+        new Object[][] {
+          {
+            singletonList(new Filter("from", "0x0000000000000000000000000000000000000001", EQ)),
+            100,
+            singletonList("1")
+          },
+          {
+            singletonList(new Filter("to", "0x0000000000000000000000000000000000000002", EQ)),
+            100,
+            singletonList("1")
+          },
+          {singletonList(new Filter("gas", "0x01", EQ)), 100, singletonList("1")},
+          {singletonList(new Filter("gas", "0x01", LT)), 100, EMPTY_LIST},
+          {singletonList(new Filter("gas", "0x01", GT)), 100, asList("2", "3", "4")},
+          {singletonList(new Filter("gas", "0x01", GT)), 1, singletonList("2")},
+          {singletonList(new Filter("gasPrice", "0x01", EQ)), 100, singletonList("1")},
+          {singletonList(new Filter("gasPrice", "0x01", LT)), 100, EMPTY_LIST},
+          {singletonList(new Filter("gasPrice", "0x01", GT)), 100, asList("2", "3", "4")},
+          {singletonList(new Filter("gasPrice", "0x01", GT)), 1, singletonList("2")},
+          {singletonList(new Filter("value", "0x01", EQ)), 100, singletonList("1")},
+          {singletonList(new Filter("value", "0x01", LT)), 100, EMPTY_LIST},
+          {singletonList(new Filter("value", "0x01", GT)), 100, asList("2", "3", "4")},
+          {singletonList(new Filter("value", "0x01", GT)), 1, singletonList("2")},
+          {singletonList(new Filter("nonce", "0x01", EQ)), 100, singletonList("1")},
+          {singletonList(new Filter("nonce", "0x01", LT)), 100, EMPTY_LIST},
+          {singletonList(new Filter("nonce", "0x01", GT)), 100, asList("2", "3", "4")},
+          {singletonList(new Filter("nonce", "0x01", GT)), 1, singletonList("2")},
+          {
+            asList(new Filter("gas", "0x03", GT), new Filter("gasPrice", "0x02", GT)),
+            100,
+            singletonList("4")
+          },
+          {
+            asList(new Filter("from", "0x01", EQ), new Filter("gasPrice", "0x02", GT)),
+            100,
+            EMPTY_LIST
+          },
+          {singletonList(new Filter("to", "contract_creation", ACTION)), 1, singletonList("4")},
+        });
+  }
+
+  private final PendingTransactionFilter pendingTransactionFilter = new PendingTransactionFilter();
+
+  private final List<Filter> filters;
+  private final int limit;
+  private final List<String> expectedListOfTransactionHash;
+
+  public PendingTransactionFilterTest(
+      final List<Filter> filters,
+      final int limit,
+      final List<String> expectedListOfTransactionHash) {
+    this.filters = filters;
+    this.limit = limit;
+    this.expectedListOfTransactionHash =
+        expectedListOfTransactionHash.stream()
+            .map(Hash::fromHexStringLenient)
+            .map(Hash::toHexString)
+            .collect(Collectors.toList());
+  }
+
+  @Test
+  public void localAndRemoteAddressShouldNotStartWithForwardSlash() {
+
+    final Set<Transaction> filteredList =
+        pendingTransactionFilter.reduce(getPendingTransactions(), filters, limit);
+
+    assertThat(filteredList.size()).isEqualTo(expectedListOfTransactionHash.size());
+    for (Transaction trx : filteredList) {
+      assertThat(expectedListOfTransactionHash).contains(trx.getHash().toHexString());
+    }
+  }
+
+  private Set<PendingTransactions.TransactionInfo> getPendingTransactions() {
+    final List<PendingTransactions.TransactionInfo> transactionInfoList = new ArrayList<>();
+    final int numberTrx = 5;
+    for (int i = 1; i < numberTrx; i++) {
+      Transaction transaction = mock(Transaction.class);
+      when(transaction.getGasPrice()).thenReturn(Wei.of(i));
+      when(transaction.getValue()).thenReturn(Wei.of(i));
+      when(transaction.getGasLimit()).thenReturn((long) i);
+      when(transaction.getNonce()).thenReturn((long) i);
+      when(transaction.getSender()).thenReturn(Address.fromHexString(String.valueOf(i)));
+      when(transaction.getTo())
+          .thenReturn(Optional.of(Address.fromHexString(String.valueOf(i + 1))));
+      when(transaction.getHash()).thenReturn(Hash.fromHexStringLenient(String.valueOf(i)));
+      if (i == numberTrx - 1) {
+        when(transaction.isContractCreation()).thenReturn(true);
+      }
+      transactionInfoList.add(
+          new PendingTransactions.TransactionInfo(
+              transaction, true, Instant.ofEpochSecond(Integer.MAX_VALUE)));
+    }
+    return new LinkedHashSet<>(transactionInfoList);
+  }
+}