Listen to tx pre event and trigger 'pending'

core/filter.go

@@ -34,7 +34,7 @@ type Filter struct {
 	topics   [][][]byte
 
 	BlockCallback   func(*types.Block, state.Logs)
-	PendingCallback func(*types.Block, state.Logs)
+	PendingCallback func(*types.Transaction)
 	LogsCallback    func(state.Logs)
 }
 

event/filter/eth_filter.go

@@ -62,8 +62,9 @@ func (self *FilterManager) GetFilter(id int) *core.Filter {
 func (self *FilterManager) filterLoop() {
 	// Subscribe to events
 	events := self.eventMux.Subscribe(
-		core.PendingBlockEvent{},
+		//core.PendingBlockEvent{},
 		core.ChainEvent{},
+		core.TxPreEvent{},
 		state.Logs(nil))
 
 out:
@@ -82,11 +83,11 @@ out:
 				}
 				self.filterMu.RUnlock()
 
-			case core.PendingBlockEvent:
+			case core.TxPreEvent:
 				self.filterMu.RLock()
 				for _, filter := range self.filters {
 					if filter.PendingCallback != nil {
-						filter.PendingCallback(event.Block, event.Logs)
+						filter.PendingCallback(event.Tx)
 					}
 				}
 				self.filterMu.RUnlock()

rpc/api.go

@@ -180,7 +180,16 @@ func (self *EthereumApi) NewFilterString(args *FilterStringArgs, reply *interfac
 	var id int
 	filter := core.NewFilter(self.xeth().Backend())
 
-	callback := func(block *types.Block, logs state.Logs) {
+	switch args.Word {
+	case "pending":
+		filter.PendingCallback = func(tx *types.Transaction) {
+			self.logMut.Lock()
+			defer self.logMut.Unlock()
+
+			self.logs[id].add(&state.StateLog{})
+		}
+	case "latest":
+		filter.BlockCallback = func(block *types.Block, logs state.Logs) {
 			self.logMut.Lock()
 			defer self.logMut.Unlock()
 
@@ -189,12 +198,6 @@ func (self *EthereumApi) NewFilterString(args *FilterStringArgs, reply *interfac
 			}
 			self.logs[id].add(&state.StateLog{})
 		}
-
-	switch args.Word {
-	case "pending":
-		filter.PendingCallback = callback
-	case "latest":
-		filter.BlockCallback = callback
 	default:
 		return NewValidationError("Word", "Must be `latest` or `pending`")
 	}