From 4eab5b821308d51dd9f6724b4e7233c4f4d4a05c Mon Sep 17 00:00:00 2001
From: Austin Roberts <code@ausiv.com>
Date: Wed, 8 Apr 2020 11:37:48 -0500
Subject: [PATCH] eth/filters, accounts/abi/bind/backends: Support websocket
 subscriptions of dropped and rejected txs

This adds the final piece for being able to get a websocket subscription to the new
dropped and rejected transactions feeds. An update was also needed in the abi/bind
simulated backend for it to comply with the newly specified interfaces.
---
 .../bind/backends/dropped_tx_subscription.go  | 16 ++++
 eth/filters/dropped_tx_subscription.go        | 82 +++++++++++++++++++
 eth/filters/filter.go                         |  2 +
 3 files changed, 100 insertions(+)
 create mode 100644 accounts/abi/bind/backends/dropped_tx_subscription.go
 create mode 100644 eth/filters/dropped_tx_subscription.go

diff --git a/accounts/abi/bind/backends/dropped_tx_subscription.go b/accounts/abi/bind/backends/dropped_tx_subscription.go
new file mode 100644
index 0000000000..853d8ece6d
--- /dev/null
+++ b/accounts/abi/bind/backends/dropped_tx_subscription.go
@@ -0,0 +1,16 @@
+package backends
+
+
+import (
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/event"
+)
+
+
+func (fb *filterBackend) SubscribeDropTxsEvent(ch chan<- core.DropTxsEvent) event.Subscription {
+	return nullSubscription()
+}
+
+func (fb *filterBackend) SubscribeRejectedTxEvent(ch chan<- core.RejectedTxEvent) event.Subscription {
+	return nullSubscription()
+}
diff --git a/eth/filters/dropped_tx_subscription.go b/eth/filters/dropped_tx_subscription.go
new file mode 100644
index 0000000000..5e95c0c8e9
--- /dev/null
+++ b/eth/filters/dropped_tx_subscription.go
@@ -0,0 +1,82 @@
+package filters
+
+import (
+	"context"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rpc"
+
+)
+
+type dropNotification struct {
+	TxHash common.Hash `json:"txhash"`
+	Reason string `json:"reason"`
+}
+
+type rejectNotification struct {
+	Tx *types.Transaction
+	Reason string `json:"reason"`
+}
+
+// DroppedTransactions send a notification each time a transaction is dropped from the mempool
+func (api *PublicFilterAPI) DroppedTransactions(ctx context.Context) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
+
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		dropped := make(chan core.DropTxsEvent)
+		droppedSub := api.backend.SubscribeDropTxsEvent(dropped)
+
+		for {
+			select {
+			case d := <-dropped:
+				for _, tx := range d.Txs {
+					notifier.Notify(rpcSub.ID, &dropNotification{TxHash: tx.Hash(), Reason: d.Reason})
+				}
+			case <-rpcSub.Err():
+				droppedSub.Unsubscribe()
+				return
+			case <-notifier.Closed():
+				droppedSub.Unsubscribe()
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
+// RejectedTransactions send a notification each time a transaction is rejected from entering the mempool
+func (api *PublicFilterAPI) RejectedTransactions(ctx context.Context) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
+
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		rejected := make(chan core.RejectedTxEvent)
+		rejectedSub := api.backend.SubscribeRejectedTxEvent(rejected)
+
+		for {
+			select {
+			case d := <-rejected:
+				notifier.Notify(rpcSub.ID, &rejectNotification{Tx: d.Tx, Reason: d.Reason.Error()})
+			case <-rpcSub.Err():
+				rejectedSub.Unsubscribe()
+				return
+			case <-notifier.Closed():
+				rejectedSub.Unsubscribe()
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 2184092071..a2a55bd0d2 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -38,6 +38,8 @@ type Backend interface {
 	GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
 
 	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
+	SubscribeDropTxsEvent(ch chan<- core.DropTxsEvent) event.Subscription
+	SubscribeRejectedTxEvent(ch chan<- core.RejectedTxEvent) event.Subscription
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
 	SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
 	SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription