rpc, ui/qt/qwhisper, whisper, xeth: introduce complex topic filters

rpc/api.go

@@ -422,12 +422,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
 
 	case "shh_newIdentity":
 		*reply = api.xeth().Whisper().NewIdentity()
-	// case "shh_removeIdentity":
-	// 	args := new(WhisperIdentityArgs)
-	// 	if err := json.Unmarshal(req.Params, &args); err != nil {
-	// 		return err
-	// 	}
-	// 	*reply = api.xeth().Whisper().RemoveIdentity(args.Identity)
+
 	case "shh_hasIdentity":
 		args := new(WhisperIdentityArgs)
 		if err := json.Unmarshal(req.Params, &args); err != nil {
@@ -439,6 +434,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
 		return NewNotImplementedError(req.Method)
 
 	case "shh_newFilter":
+		// Create a new filter to watch and match messages with
 		args := new(WhisperFilterArgs)
 		if err := json.Unmarshal(req.Params, &args); err != nil {
 			return err

rpc/args.go

@@ -1010,25 +1010,27 @@ func (args *WhisperIdentityArgs) UnmarshalJSON(b []byte) (err error) {
 }
 
 type WhisperFilterArgs struct {
-	To     string `json:"to"`
+	To     string
 	From   string
-	Topics []string
+	Topics [][]string
 }
 
+// UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a
+// JSON message blob into a WhisperFilterArgs structure.
 func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) {
+	// Unmarshal the JSON message and sanity check
 	var obj []struct {
-		To     interface{}
-		Topics []interface{}
+		To     interface{} `json:"to"`
+		From   interface{} `json:"from"`
+		Topics interface{} `json:"topics"`
 	}
-
-	if err = json.Unmarshal(b, &obj); err != nil {
+	if err := json.Unmarshal(b, &obj); err != nil {
 		return NewDecodeParamError(err.Error())
 	}
-
 	if len(obj) < 1 {
 		return NewInsufficientParamsError(len(obj), 1)
 	}
-
+	// Retrieve the simple data contents of the filter arguments
 	if obj[0].To == nil {
 		args.To = ""
 	} else {
@@ -1038,17 +1040,52 @@ func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) {
 		}
 		args.To = argstr
 	}
-
-	t := make([]string, len(obj[0].Topics))
-	for i, j := range obj[0].Topics {
-		argstr, ok := j.(string)
+	if obj[0].From == nil {
+		args.From = ""
+	} else {
+		argstr, ok := obj[0].From.(string)
 		if !ok {
-			return NewInvalidTypeError("topics["+string(i)+"]", "is not a string")
+			return NewInvalidTypeError("from", "is not a string")
+		}
+		args.From = argstr
 	}
-		t[i] = argstr
+	// Construct the nested topic array
+	if obj[0].Topics != nil {
+		// Make sure we have an actual topic array
+		list, ok := obj[0].Topics.([]interface{})
+		if !ok {
+			return NewInvalidTypeError("topics", "is not an array")
 		}
-	args.Topics = t
+		// Iterate over each topic and handle nil, string or array
+		topics := make([][]string, len(list))
+		for idx, field := range list {
+			switch value := field.(type) {
+			case nil:
+				topics[idx] = []string{""}
+
+			case string:
+				topics[idx] = []string{value}
 
+			case []interface{}:
+				topics[idx] = make([]string, len(value))
+				for i, nested := range value {
+					switch value := nested.(type) {
+					case nil:
+						topics[idx][i] = ""
+
+					case string:
+						topics[idx][i] = value
+
+					default:
+						return NewInvalidTypeError(fmt.Sprintf("topic[%d][%d]", idx, i), "is not a string")
+					}
+				}
+			default:
+				return NewInvalidTypeError(fmt.Sprintf("topic[%d]", idx), "not a string or array")
+			}
+		}
+		args.Topics = topics
+	}
 	return nil
 }
 

rpc/args_test.go

@@ -1943,7 +1943,7 @@ func TestWhisperFilterArgs(t *testing.T) {
 	input := `[{"topics": ["0x68656c6c6f20776f726c64"], "to": "0x34ag445g3455b34"}]`
 	expected := new(WhisperFilterArgs)
 	expected.To = "0x34ag445g3455b34"
-	expected.Topics = []string{"0x68656c6c6f20776f726c64"}
+	expected.Topics = [][]string{[]string{"0x68656c6c6f20776f726c64"}}
 
 	args := new(WhisperFilterArgs)
 	if err := json.Unmarshal([]byte(input), &args); err != nil {

ui/qt/qwhisper/whisper.go

@@ -106,7 +106,7 @@ func filterFromMap(opts map[string]interface{}) (f whisper.Filter) {
 	if topicList, ok := opts["topics"].(*qml.List); ok {
 		var topics []string
 		topicList.Convert(&topics)
-		f.Topics = whisper.NewTopicsFromStrings(topics...)
+		f.Topics = whisper.NewTopicFilterFromStringsFlat(topics...)
 	}
 
 	return

whisper/filter.go

@@ -2,12 +2,55 @@
 
 package whisper
 
-import "crypto/ecdsa"
+import (
+	"crypto/ecdsa"
+
+	"github.com/ethereum/go-ethereum/event/filter"
+)
 
 // Filter is used to subscribe to specific types of whisper messages.
 type Filter struct {
 	To     *ecdsa.PublicKey   // Recipient of the message
 	From   *ecdsa.PublicKey   // Sender of the message
-	Topics []Topic          // Topics to watch messages on
-	Fn     func(*Message)   // Handler in case of a match
+	Topics [][]Topic          // Topics to filter messages with
+	Fn     func(msg *Message) // Handler in case of a match
+}
+
+// filterer is the internal, fully initialized filter ready to match inbound
+// messages to a variety of criteria.
+type filterer struct {
+	to      string                 // Recipient of the message
+	from    string                 // Sender of the message
+	matcher *topicMatcher          // Topics to filter messages with
+	fn      func(data interface{}) // Handler in case of a match
+}
+
+// Compare checks if the specified filter matches the current one.
+func (self filterer) Compare(f filter.Filter) bool {
+	filter := f.(filterer)
+
+	// Check the message sender and recipient
+	if len(self.to) > 0 && self.to != filter.to {
+		return false
+	}
+	if len(self.from) > 0 && self.from != filter.from {
+		return false
+	}
+	// Check the topic filtering
+	topics := make([]Topic, len(filter.matcher.conditions))
+	for i, group := range filter.matcher.conditions {
+		// Message should contain a single topic entry, extract
+		for topics[i], _ = range group {
+			break
+		}
+	}
+	if !self.matcher.Matches(topics) {
+		return false
+	}
+	return true
+}
+
+// Trigger is called when a filter successfully matches an inbound message.
+func (self filterer) Trigger(data interface{}) {
+	self.fn(data)
 }

whisper/topic.go

@@ -27,6 +27,26 @@ func NewTopics(data ...[]byte) []Topic {
 	return topics
 }
 
+// NewTopicFilter creates a 2D topic array used by whisper.Filter from binary
+// data elements.
+func NewTopicFilter(data ...[][]byte) [][]Topic {
+	filter := make([][]Topic, len(data))
+	for i, condition := range data {
+		filter[i] = NewTopics(condition...)
+	}
+	return filter
+}
+
+// NewTopicFilterFlat creates a 2D topic array used by whisper.Filter from flat
+// binary data elements.
+func NewTopicFilterFlat(data ...[]byte) [][]Topic {
+	filter := make([][]Topic, len(data))
+	for i, element := range data {
+		filter[i] = []Topic{NewTopic(element)}
+	}
+	return filter
+}
+
 // NewTopicFromString creates a topic using the binary data contents of the
 // specified string.
 func NewTopicFromString(data string) Topic {
@@ -43,19 +63,100 @@ func NewTopicsFromStrings(data ...string) []Topic {
 	return topics
 }
 
+// NewTopicFilterFromStrings creates a 2D topic array used by whisper.Filter
+// from textual data elements.
+func NewTopicFilterFromStrings(data ...[]string) [][]Topic {
+	filter := make([][]Topic, len(data))
+	for i, condition := range data {
+		filter[i] = NewTopicsFromStrings(condition...)
+	}
+	return filter
+}
+
+// NewTopicFilterFromStringsFlat creates a 2D topic array used by whisper.Filter from flat
+// binary data elements.
+func NewTopicFilterFromStringsFlat(data ...string) [][]Topic {
+	filter := make([][]Topic, len(data))
+	for i, element := range data {
+		filter[i] = []Topic{NewTopicFromString(element)}
+	}
+	return filter
+}
+
 // String converts a topic byte array to a string representation.
 func (self *Topic) String() string {
 	return string(self[:])
 }
 
-// TopicSet represents a hash set to check if a topic exists or not.
-type topicSet map[string]struct{}
+// topicMatcher is a filter expression to verify if a list of topics contained
+// in an arriving message matches some topic conditions. The topic matcher is
+// built up of a list of conditions, each of which must be satisfied by the
+// corresponding topic in the message. Each condition may require: a) an exact
+// topic match; b) a match from a set of topics; or c) a wild-card matching all.
+//
+// If a message contains more topics than required by the matcher, those beyond
+// the condition count are ignored and assumed to match.
+//
+// Consider the following sample topic matcher:
+//   sample := {
+//     {TopicA1, TopicA2, TopicA3},
+//     {TopicB},
+//     nil,
+//     {TopicD1, TopicD2}
+//   }
+// In order for a message to pass this filter, it should enumerate at least 4
+// topics, the first any of [TopicA1, TopicA2, TopicA3], the second mandatory
+// "TopicB", the third is ignored by the filter and the fourth either "TopicD1"
+// or "TopicD2". If the message contains further topics, the filter will match
+// them too.
+type topicMatcher struct {
+	conditions []map[Topic]struct{}
+}
+
+// newTopicMatcher create a topic matcher from a list of topic conditions.
+func newTopicMatcher(topics ...[]Topic) *topicMatcher {
+	matcher := make([]map[Topic]struct{}, len(topics))
+	for i, condition := range topics {
+		matcher[i] = make(map[Topic]struct{})
+		for _, topic := range condition {
+			matcher[i][topic] = struct{}{}
+		}
+	}
+	return &topicMatcher{conditions: matcher}
+}
+
+// newTopicMatcherFromBinary create a topic matcher from a list of binary conditions.
+func newTopicMatcherFromBinary(data ...[][]byte) *topicMatcher {
+	topics := make([][]Topic, len(data))
+	for i, condition := range data {
+		topics[i] = NewTopics(condition...)
+	}
+	return newTopicMatcher(topics...)
+}
+
+// newTopicMatcherFromStrings creates a topic matcher from a list of textual
+// conditions.
+func newTopicMatcherFromStrings(data ...[]string) *topicMatcher {
+	topics := make([][]Topic, len(data))
+	for i, condition := range data {
+		topics[i] = NewTopicsFromStrings(condition...)
+	}
+	return newTopicMatcher(topics...)
+}
 
-// NewTopicSet creates a topic hash set from a slice of topics.
-func newTopicSet(topics []Topic) topicSet {
-	set := make(map[string]struct{})
-	for _, topic := range topics {
-		set[topic.String()] = struct{}{}
+// Matches checks if a list of topics matches this particular condition set.
+func (self *topicMatcher) Matches(topics []Topic) bool {
+	// Mismatch if there aren't enough topics
+	if len(self.conditions) > len(topics) {
+		return false
+	}
+	// Check each topic condition for existence (skip wild-cards)
+	for i := 0; i < len(topics) && i < len(self.conditions); i++ {
+		if len(self.conditions[i]) > 0 {
+			if _, ok := self.conditions[i][topics[i]]; !ok {
+				return false
+			}
+		}
 	}
-	return topicSet(set)
+	return true
 }

whisper/topic_test.go

@@ -52,16 +52,149 @@ func TestTopicCreation(t *testing.T) {
 	}
 }
 
-func TestTopicSetCreation(t *testing.T) {
-	topics := make([]Topic, len(topicCreationTests))
-	for i, tt := range topicCreationTests {
-		topics[i] = NewTopic(tt.data)
+var topicMatcherCreationTest = struct {
+	binary  [][][]byte
+	textual [][]string
+	matcher []map[[4]byte]struct{}
+}{
+	binary: [][][]byte{
+		[][]byte{},
+		[][]byte{
+			[]byte("Topic A"),
+		},
+		[][]byte{
+			[]byte("Topic B1"),
+			[]byte("Topic B2"),
+			[]byte("Topic B3"),
+		},
+	},
+	textual: [][]string{
+		[]string{},
+		[]string{"Topic A"},
+		[]string{"Topic B1", "Topic B2", "Topic B3"},
+	},
+	matcher: []map[[4]byte]struct{}{
+		map[[4]byte]struct{}{},
+		map[[4]byte]struct{}{
+			[4]byte{0x25, 0xfc, 0x95, 0x66}: struct{}{},
+		},
+		map[[4]byte]struct{}{
+			[4]byte{0x93, 0x6d, 0xec, 0x09}: struct{}{},
+			[4]byte{0x25, 0x23, 0x34, 0xd3}: struct{}{},
+			[4]byte{0x6b, 0xc2, 0x73, 0xd1}: struct{}{},
+		},
+	},
 }
-	set := newTopicSet(topics)
-	for i, tt := range topicCreationTests {
-		topic := NewTopic(tt.data)
-		if _, ok := set[topic.String()]; !ok {
-			t.Errorf("topic %d: not found in set", i)
+
+func TestTopicMatcherCreation(t *testing.T) {
+	test := topicMatcherCreationTest
+
+	matcher := newTopicMatcherFromBinary(test.binary...)
+	for i, cond := range matcher.conditions {
+		for topic, _ := range cond {
+			if _, ok := test.matcher[i][topic]; !ok {
+				t.Errorf("condition %d; extra topic found: 0x%x", i, topic[:])
+			}
+		}
+	}
+	for i, cond := range test.matcher {
+		for topic, _ := range cond {
+			if _, ok := matcher.conditions[i][topic]; !ok {
+				t.Errorf("condition %d; topic not found: 0x%x", i, topic[:])
+			}
+		}
+	}
+
+	matcher = newTopicMatcherFromStrings(test.textual...)
+	for i, cond := range matcher.conditions {
+		for topic, _ := range cond {
+			if _, ok := test.matcher[i][topic]; !ok {
+				t.Errorf("condition %d; extra topic found: 0x%x", i, topic[:])
+			}
+		}
+	}
+	for i, cond := range test.matcher {
+		for topic, _ := range cond {
+			if _, ok := matcher.conditions[i][topic]; !ok {
+				t.Errorf("condition %d; topic not found: 0x%x", i, topic[:])
+			}
+		}
+	}
+}
+
+var topicMatcherTests = []struct {
+	filter [][]string
+	topics []string
+	match  bool
+}{
+	// Empty topic matcher should match everything
+	{
+		filter: [][]string{},
+		topics: []string{},
+		match:  true,
+	},
+	{
+		filter: [][]string{},
+		topics: []string{"a", "b", "c"},
+		match:  true,
+	},
+	// Fixed topic matcher should match strictly, but only prefix
+	{
+		filter: [][]string{[]string{"a"}, []string{"b"}},
+		topics: []string{"a"},
+		match:  false,
+	},
+	{
+		filter: [][]string{[]string{"a"}, []string{"b"}},
+		topics: []string{"a", "b"},
+		match:  true,
+	},
+	{
+		filter: [][]string{[]string{"a"}, []string{"b"}},
+		topics: []string{"a", "b", "c"},
+		match:  true,
+	},
+	// Multi-matcher should match any from a sub-group
+	{
+		filter: [][]string{[]string{"a1", "a2"}},
+		topics: []string{"a"},
+		match:  false,
+	},
+	{
+		filter: [][]string{[]string{"a1", "a2"}},
+		topics: []string{"a1"},
+		match:  true,
+	},
+	{
+		filter: [][]string{[]string{"a1", "a2"}},
+		topics: []string{"a2"},
+		match:  true,
+	},
+	// Wild-card condition should match anything
+	{
+		filter: [][]string{[]string{}, []string{"b"}},
+		topics: []string{"a"},
+		match:  false,
+	},
+	{
+		filter: [][]string{[]string{}, []string{"b"}},
+		topics: []string{"a", "b"},
+		match:  true,
+	},
+	{
+		filter: [][]string{[]string{}, []string{"b"}},
+		topics: []string{"b", "b"},
+		match:  true,
+	},
+}
+
+func TestTopicMatcher(t *testing.T) {
+	for i, tt := range topicMatcherTests {
+		topics := NewTopicsFromStrings(tt.topics...)
+
+		matcher := newTopicMatcherFromStrings(tt.filter...)
+		if match := matcher.Matches(topics); match != tt.match {
+			t.Errorf("test %d: match mismatch: have %v, want %v", i, match, tt.match)
 		}
 	}
 }

whisper/whisper.go

@@ -118,11 +118,11 @@ func (self *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey {
 // Watch installs a new message handler to run in case a matching packet arrives
 // from the whisper network.
 func (self *Whisper) Watch(options Filter) int {
-	filter := filter.Generic{
-		Str1: string(crypto.FromECDSAPub(options.To)),
-		Str2: string(crypto.FromECDSAPub(options.From)),
-		Data: newTopicSet(options.Topics),
-		Fn: func(data interface{}) {
+	filter := filterer{
+		to:      string(crypto.FromECDSAPub(options.To)),
+		from:    string(crypto.FromECDSAPub(options.From)),
+		matcher: newTopicMatcher(options.Topics...),
+		fn: func(data interface{}) {
 			options.Fn(data.(*Message))
 		},
 	}
@@ -273,10 +273,14 @@ func (self *Whisper) open(envelope *Envelope) *Message {
 
 // createFilter creates a message filter to check against installed handlers.
 func createFilter(message *Message, topics []Topic) filter.Filter {
-	return filter.Generic{
-		Str1: string(crypto.FromECDSAPub(message.To)),
-		Str2: string(crypto.FromECDSAPub(message.Recover())),
-		Data: newTopicSet(topics),
+	matcher := make([][]Topic, len(topics))
+	for i, topic := range topics {
+		matcher[i] = []Topic{topic}
+	}
+	return filterer{
+		to:      string(crypto.FromECDSAPub(message.To)),
+		from:    string(crypto.FromECDSAPub(message.Recover())),
+		matcher: newTopicMatcher(matcher...),
 	}
 }
 

whisper/whisper_test.go

@@ -129,7 +129,7 @@ func testBroadcast(anonymous bool, t *testing.T) {
 		dones[i] = done
 
 		targets[i].Watch(Filter{
-			Topics: NewTopicsFromStrings("broadcast topic"),
+			Topics: NewTopicFilterFromStrings([]string{"broadcast topic"}),
 			Fn: func(msg *Message) {
 				close(done)
 			},

xeth/whisper.go

@@ -67,11 +67,11 @@ func (self *Whisper) Post(payload string, to, from string, topics []string, prio
 
 // Watch installs a new message handler to run in case a matching packet arrives
 // from the whisper network.
-func (self *Whisper) Watch(to, from string, topics []string, fn func(WhisperMessage)) int {
+func (self *Whisper) Watch(to, from string, topics [][]string, fn func(WhisperMessage)) int {
 	filter := whisper.Filter{
 		To:     crypto.ToECDSAPub(common.FromHex(to)),
 		From:   crypto.ToECDSAPub(common.FromHex(from)),
-		Topics: whisper.NewTopicsFromStrings(topics...),
+		Topics: whisper.NewTopicFilterFromStrings(topics...),
 	}
 	filter.Fn = func(message *whisper.Message) {
 		fn(NewWhisperMessage(message))

xeth/xeth.go

@@ -452,7 +452,7 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin
 	return filter.Find()
 }
 
-func (p *XEth) NewWhisperFilter(to, from string, topics []string) int {
+func (p *XEth) NewWhisperFilter(to, from string, topics [][]string) int {
 	var id int
 	callback := func(msg WhisperMessage) {
 		p.messagesMut.Lock()