Merge 37f0f2c824 into 7843192c8e

whisper/whisperv5/filter.go

@@ -26,6 +26,11 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const (
+	ALL_TOPICS        = ""
+	MAX_POOL_CAPACITY = 1000
+)
+
 type Filter struct {
 	Src        *ecdsa.PublicKey  // Sender of the message
 	KeyAsym    *ecdsa.PrivateKey // Private Key of recipient
@@ -40,16 +45,19 @@ type Filter struct {
 }
 
 type Filters struct {
-	watchers map[string]*Filter
+	watchers     map[string]*Filter
-	whisper  *Whisper
+	whisper      *Whisper
-	mutex    sync.RWMutex
+	mutex        sync.RWMutex
+	topicMatcher *topicMatcher
 }
 
 func NewFilters(w *Whisper) *Filters {
-	return &Filters{
+	fs := &Filters{
-		watchers: make(map[string]*Filter),
+		watchers:     make(map[string]*Filter),
-		whisper:  w,
+		whisper:      w,
+		topicMatcher: newTopicMatcher(),
 	}
+	return fs
 }
 
 func (fs *Filters) Install(watcher *Filter) (string, error) {
@@ -74,6 +82,7 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 	}
 
 	fs.watchers[id] = watcher
+	fs.topicMatcher.addFilterToTopicsMapping(watcher, id)
 	return id, err
 }
 
@@ -82,6 +91,7 @@ func (fs *Filters) Uninstall(id string) bool {
 	defer fs.mutex.Unlock()
 	if fs.watchers[id] != nil {
 		delete(fs.watchers, id)
+		fs.topicMatcher.removeTopicFromTopicMapping(id)
 		return true
 	}
 	return false
@@ -95,15 +105,22 @@ func (fs *Filters) Get(id string) *Filter {
 
 func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	var msg *ReceivedMessage
+	matchedTopics := fs.topicMatcher.take()
+	defer fs.topicMatcher.resolve(matchedTopics)
 
 	fs.mutex.RLock()
 	defer fs.mutex.RUnlock()
 
-	i := -1 // only used for logging info
+	fs.topicMatcher.matchedTopics(env.Topic, &matchedTopics)
-	for _, watcher := range fs.watchers {
+	for _, watcherID := range matchedTopics {
-		i++
+		watcher, ok := fs.watchers[watcherID]
+		if !ok {
+			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: filter not exists", env.Hash(), watcherID))
+			continue
+		}
+
 		if p2pMessage && !watcher.AllowP2P {
-			log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), i))
+			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcherID))
 			continue
 		}
 
@@ -115,10 +132,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 			if match {
 				msg = env.Open(watcher)
 				if msg == nil {
-					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i)
+					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcherID)
 				}
 			} else {
-				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i)
+				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcherID)
 			}
 		}
 
@@ -181,9 +198,9 @@ func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
 	}
 
 	if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
-		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) && f.MatchTopic(msg.Topic)
+		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst)
 	} else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
-		return f.SymKeyHash == msg.SymKeyHash && f.MatchTopic(msg.Topic)
+		return f.SymKeyHash == msg.SymKeyHash
 	}
 	return false
 }
@@ -194,44 +211,13 @@ func (f *Filter) MatchEnvelope(envelope *Envelope) bool {
 	}
 
 	if f.expectsAsymmetricEncryption() && envelope.isAsymmetric() {
-		return f.MatchTopic(envelope.Topic)
+		return true
 	} else if f.expectsSymmetricEncryption() && envelope.IsSymmetric() {
-		return f.MatchTopic(envelope.Topic)
-	}
-	return false
-}
-
-func (f *Filter) MatchTopic(topic TopicType) bool {
-	if len(f.Topics) == 0 {
-		// any topic matches
 		return true
 	}
-
-	for _, bt := range f.Topics {
-		if matchSingleTopic(topic, bt) {
-			return true
-		}
-	}
 	return false
 }
 
-func matchSingleTopic(topic TopicType, bt []byte) bool {
-	if len(bt) > TopicLength {
-		bt = bt[:TopicLength]
-	}
-
-	if len(bt) < TopicLength {
-		return false
-	}
-
-	for j, b := range bt {
-		if topic[j] != b {
-			return false
-		}
-	}
-	return true
-}
-
 func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
 	if !ValidatePublicKey(a) {
 		return false
@@ -241,3 +227,91 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
 	// the curve is always the same, just compare the points
 	return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0
 }
+
+//topicMatcher keeps topic->watcher mapping
+type topicMatcher struct {
+	//structure - map[topic]map[filterID]
+	//mapping for topics
+	//"" topic means that the filter allows all topic values
+	mapper map[string]map[string]struct{}
+	mx     sync.RWMutex
+	pool   sync.Pool
+}
+
+//newTopicMatcher returns a newly created topic matcher
+func newTopicMatcher() *topicMatcher {
+	tm := new(topicMatcher)
+	tm.mapper = make(map[string]map[string]struct{})
+	tm.mapper[ALL_TOPICS] = make(map[string]struct{})
+	tm.pool.New = func() interface{} {
+		return []string{}
+	}
+	return tm
+}
+
+//take returns []string from pool
+func (fs *topicMatcher) take() []string {
+	return fs.pool.Get().([]string)
+}
+
+//resolve put []string to pool
+func (fs *topicMatcher) resolve(s []string) {
+	if cap(s) > MAX_POOL_CAPACITY {
+		return
+	}
+	fs.pool.Put(s[:0])
+}
+
+//addFilterToTopicsMapping fill topic->watcher mapping for current watcher
+func (fs *topicMatcher) addFilterToTopicsMapping(watcher *Filter, id string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+
+	for i := range fs.prepareTopicsMapping(watcher) {
+		topicMapping, ok := fs.mapper[i]
+		if !ok {
+			fs.mapper[i] = make(map[string]struct{})
+			topicMapping = fs.mapper[i]
+		}
+		topicMapping[id] = struct{}{}
+	}
+}
+
+//removeTopicFromTopicMapping removes mapping info by filterID
+func (fs *topicMatcher) removeTopicFromTopicMapping(id string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+	for i := range fs.mapper {
+		delete(fs.mapper[i], id)
+	}
+}
+
+//prepareTopicsMapping returns set of topics for watcher
+func (fs *topicMatcher) prepareTopicsMapping(watcher *Filter) map[string]struct{} {
+	topics := make(map[string]struct{}, len(watcher.Topics))
+
+	if len(watcher.Topics) == 0 {
+		topics[ALL_TOPICS] = struct{}{}
+		return topics
+	}
+
+	for _, topic := range watcher.Topics {
+		topics[common.ToHex(topic)] = struct{}{}
+	}
+
+	return topics
+}
+
+//matchedTopics write all matched topics to matched
+func (fs *topicMatcher) matchedTopics(topic TopicType, matched *[]string) {
+	fs.mx.RLock()
+	defer fs.mx.RUnlock()
+
+	for i := range fs.mapper[ALL_TOPICS] {
+		*matched = append(*matched, i)
+	}
+
+	for i := range fs.mapper[topic.String()] {
+		*matched = append(*matched, i)
+	}
+}

whisper/whisperv5/filter_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"fmt"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 )
@@ -48,7 +49,7 @@ type FilterTestCase struct {
 	msgCnt int
 }
 
-func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
+func generateFilter(symmetric bool) (*Filter, error) {
 	var f Filter
 	f.Messages = make(map[common.Hash]*ReceivedMessage)
 
@@ -62,8 +63,8 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 
 	key, err := crypto.GenerateKey()
 	if err != nil {
-		t.Fatalf("generateFilter 1 failed with seed %d.", seed)
+		return nil, fmt.Errorf("generateFilter 1 failed with seed %d. Error: %s", seed, err.Error())
-		return nil, err
+
 	}
 	f.Src = &key.PublicKey
 
@@ -74,8 +75,7 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 	} else {
 		f.KeyAsym, err = crypto.GenerateKey()
 		if err != nil {
-			t.Fatalf("generateFilter 2 failed with seed %d.", seed)
+			return nil, fmt.Errorf("generateFilter 2 failed with seed %d. Error: %s", seed, err.Error())
-			return nil, err
 		}
 	}
 
@@ -83,10 +83,21 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 	return &f, nil
 }
 
+func generateFilters() *Filters {
+	fs := Filters{
+		watchers:     make(map[string]*Filter),
+		topicMatcher: newTopicMatcher(),
+	}
+	return &fs
+}
+
 func generateTestCases(t *testing.T, SizeTestFilters int) []FilterTestCase {
 	cases := make([]FilterTestCase, SizeTestFilters)
 	for i := 0; i < SizeTestFilters; i++ {
-		f, _ := generateFilter(t, true)
+		f, err := generateFilter(true)
+		if err != nil {
+			t.Fatal(err)
+		}
 		cases[i].f = f
 		cases[i].alive = mrand.Int()&int(1) == 0
 	}
@@ -137,8 +148,10 @@ func TestInstallSymKeyGeneratesHash(t *testing.T) {
 
 	w := New(&Config{})
 	filters := NewFilters(w)
-	filter, _ := generateFilter(t, true)
+	filter, err := generateFilter(true)
-
+	if err != nil {
+		t.Fatal(err)
+	}
 	// save the current SymKeyHash for comparison
 	initialSymKeyHash := filter.SymKeyHash
 
@@ -146,7 +159,7 @@ func TestInstallSymKeyGeneratesHash(t *testing.T) {
 	var invalid common.Hash
 	filter.SymKeyHash = invalid
 
-	_, err := filters.Install(filter)
+	_, err = filters.Install(filter)
 
 	if err != nil {
 		t.Fatalf("Error installing the filter: %s", err)
@@ -164,8 +177,10 @@ func TestInstallIdenticalFilters(t *testing.T) {
 
 	w := New(&Config{})
 	filters := NewFilters(w)
-	filter1, _ := generateFilter(t, true)
+	filter1, err := generateFilter(true)
-
+	if err != nil {
+		t.Fatal(err)
+	}
 	// Copy the first filter since some of its fields
 	// are randomly gnerated.
 	filter2 := &Filter{
@@ -176,7 +191,7 @@ func TestInstallIdenticalFilters(t *testing.T) {
 		Messages: make(map[common.Hash]*ReceivedMessage),
 	}
 
-	_, err := filters.Install(filter1)
+	_, err = filters.Install(filter1)
 
 	if err != nil {
 		t.Fatalf("Error installing the first filter with seed %d: %s", seed, err)
@@ -258,12 +273,12 @@ func TestComparePubKey(t *testing.T) {
 func TestMatchEnvelope(t *testing.T) {
 	InitSingleTest()
 
-	fsym, err := generateFilter(t, true)
+	fsym, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
 
-	fasym, err := generateFilter(t, false)
+	fasym, err := generateFilter(false)
 	if err != nil {
 		t.Fatalf("failed generateFilter() with seed %d: %s.", seed, err)
 	}
@@ -284,19 +299,7 @@ func TestMatchEnvelope(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
 	}
-	match := fsym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope symmetric with seed %d.", seed)
-	}
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope asymmetric with seed %d.", seed)
-	}
 
-	// encrypt symmetrically
-	i := mrand.Int() % 4
-	fsym.Topics[i] = params.Topic[:]
-	fasym.Topics[i] = params.Topic[:]
 	msg, err = NewSentMessage(params)
 	if err != nil {
 		t.Fatalf("failed to create new message with seed %d: %s.", seed, err)
@@ -306,41 +309,20 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed Wrap() with seed %d: %s.", seed, err)
 	}
 
-	// symmetric + matching topic: match
+	// symmetric +  insufficient PoW: mismatch
-	match = fsym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed)
-	}
-
-	// asymmetric + matching topic: mismatch
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope() asymmetric with seed %d.", seed)
-	}
-
-	// symmetric + matching topic + insufficient PoW: mismatch
 	fsym.PoW = env.PoW() + 1.0
-	match = fsym.MatchEnvelope(env)
+	match := fsym.MatchEnvelope(env)
 	if match {
 		t.Fatalf("failed MatchEnvelope(symmetric + matching topic + insufficient PoW) asymmetric with seed %d.", seed)
 	}
 
-	// symmetric + matching topic + sufficient PoW: match
+	// symmetric +  sufficient PoW: match
 	fsym.PoW = env.PoW() / 2
 	match = fsym.MatchEnvelope(env)
 	if !match {
 		t.Fatalf("failed MatchEnvelope(symmetric + matching topic + sufficient PoW) with seed %d.", seed)
 	}
 
-	// symmetric + topics are nil (wildcard): match
-	prevTopics := fsym.Topics
-	fsym.Topics = nil
-	match = fsym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(symmetric + topics are nil) with seed %d.", seed)
-	}
-	fsym.Topics = prevTopics
-
 	// encrypt asymmetrically
 	key, err := crypto.GenerateKey()
 	if err != nil {
@@ -363,26 +345,6 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(encryption method mismatch) with seed %d.", seed)
 	}
 
-	// asymmetric + mismatching topic: mismatch
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + mismatching topic) with seed %d.", seed)
-	}
-
-	// asymmetric + matching topic: match
-	fasym.Topics[i] = fasym.Topics[i+1]
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + matching topic) with seed %d.", seed)
-	}
-
-	// asymmetric + filter without topic (wildcard): match
-	fasym.Topics = nil
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + filter without topic) with seed %d.", seed)
-	}
-
 	// asymmetric + insufficient PoW: mismatch
 	fasym.PoW = env.PoW() + 1.0
 	match = fasym.MatchEnvelope(env)
@@ -397,19 +359,6 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(asymmetric + sufficient PoW) with seed %d.", seed)
 	}
 
-	// filter without topic + envelope without topic: match
-	env.Topic = TopicType{}
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed)
-	}
-
-	// filter with topic + envelope without topic: mismatch
-	fasym.Topics = fsym.Topics
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed)
-	}
 }
 
 func TestMatchMessageSym(t *testing.T) {
@@ -420,7 +369,7 @@ func TestMatchMessageSym(t *testing.T) {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
@@ -461,13 +410,6 @@ func TestMatchMessageSym(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed)
 	}
 
-	// topic mismatch
-	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
-		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
-	}
-	f.Topics[index][0]--
-
 	// key mismatch
 	f.SymKeyHash[0]++
 	if f.MatchMessage(msg) {
@@ -506,7 +448,7 @@ func TestMatchMessageSym(t *testing.T) {
 func TestMatchMessageAsym(t *testing.T) {
 	InitSingleTest()
 
-	f, err := generateFilter(t, false)
+	f, err := generateFilter(false)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
@@ -554,13 +496,6 @@ func TestMatchMessageAsym(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed)
 	}
 
-	// topic mismatch
-	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
-		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
-	}
-	f.Topics[index][0]--
-
 	// key mismatch
 	prev := *f.KeyAsym.PublicKey.X
 	zero := *big.NewInt(0)
@@ -791,58 +726,216 @@ func TestVariableTopics(t *testing.T) {
 		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
 	}
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
 
+	fs := generateFilters()
+
+	filterID, err := fs.Install(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	for i := 0; i < 4; i++ {
 		env.Topic = BytesToTopic(f.Topics[i])
+
+		//test match
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
 		match = f.MatchEnvelope(env)
-		if !match {
+		if !(match && hasFilterID(matched, filterID)) {
 			t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
 
-		f.Topics[i][lastTopicByte]++
+		//change envelop topic
+		env.Topic[lastTopicByte]++
+
+		//false positive test
 		match = f.MatchEnvelope(env)
-		if match {
+
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if match && hasFilterID(matched, filterID) {
 			t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i)
 		}
 	}
 }
 
-func TestMatchSingleTopic_ReturnTrue(t *testing.T) {
+func TestTopicsMapping(t *testing.T) {
-	bt := []byte("test")
+	InitSingleTest()
-	topic := BytesToTopic(bt)
 
-	if !matchSingleTopic(topic, bt) {
+	const lastTopicByte = 3
-		t.FailNow()
+	params, err := generateMessageParams()
+	if err != nil {
+		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
+	}
+	msg, err := NewSentMessage(params)
+	if err != nil {
+		t.Fatalf("failed to create new message with seed %d: %s.", seed, err)
+	}
+	env, err := msg.Wrap(params)
+	if err != nil {
+		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
+	}
+
+	f, err := generateFilter(true)
+	if err != nil {
+		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+	}
+	fs := generateFilters()
+
+	for i := 0; i < len(f.Topics); i++ {
+		env.Topic = BytesToTopic(f.Topics[i])
+
+		//test match
+		filterID, err := fs.Install(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if !hasFilterID(matched, filterID) {
+			t.Fatalf("failed matchedTopics, step %d.", i)
+		}
+
+		//test match without filter
+		if !fs.Uninstall(filterID) {
+			t.Fatal("Failed to uninstall filter")
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			t.Fatalf("failed match without filter, step %d.", i)
+		}
+
+		//test match with changed topic
+		f.Topics[i][lastTopicByte]++
+		filterID, err = fs.Install(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			t.Fatalf("failed match with changed topic, step %d.", i)
+		}
+		if !fs.Uninstall(filterID) {
+			t.Fatal("Failed to uninstall filter")
+		}
 	}
 }
 
-func TestMatchSingleTopic_WithTail_ReturnTrue(t *testing.T) {
+func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) {
-	bt := []byte("test with tail")
+	InitSingleTest()
-	topic := BytesToTopic([]byte("test"))
 
-	if !matchSingleTopic(topic, bt) {
+	f, err := generateFilter(true)
-		t.FailNow()
+	if err != nil {
+		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+	}
+	fs := generateFilters()
+
+	f.Topics = [][]byte{}
+
+	filterID, err := fs.Install(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	//generate topic
+	topic := TopicType{}
+	mrand.Read(topic[:])
+
+	matched := []string{}
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if !hasFilterID(matched, filterID) {
+		t.Fatal("failed matchedTopics")
+	}
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; !ok {
+		t.Fatal("watcher mapping incorrect")
+	}
+
+	//test match without filter
+	if !fs.Uninstall(filterID) {
+		t.Fatal("Failed to uninstall filter")
+	}
+	matched = matched[:0]
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if hasFilterID(matched, filterID) {
+		t.Fatal("failed match without filter")
+	}
+
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; ok {
+		t.Fatal("watcher mapping incorrect")
 	}
 }
 
-func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) {
+func hasFilterID(matched []string, filterID string) bool {
-	bt := []byte("tes")
+	for i := range matched {
-	topic := BytesToTopic(bt)
+		if matched[i] == filterID {
-
+			return true
-	if matchSingleTopic(topic, bt) {
+		}
-		t.FailNow()
 	}
+	return false
 }
 
-func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) {
+func BenchmarkFilter_MatchEnvelope_5Filters(b *testing.B) {
-	bt := []byte("test")
+	InitSingleTest()
-	topic := BytesToTopic([]byte("not_equal"))
+	benchFilter_MatchMessage(b, 5)
+}
+func BenchmarkFilter_MatchEnvelope_10Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 10)
+}
+func BenchmarkFilter_MatchEnvelope_20Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 20)
+}
+func BenchmarkFilter_MatchEnvelope_50Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 50)
+}
+func BenchmarkFilter_MatchEnvelope_100Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 100)
+}
 
-	if matchSingleTopic(topic, bt) {
+func benchFilter_MatchMessage(b *testing.B, numOfFilters int) {
-		t.FailNow()
+	params, err := generateMessageParams()
+	if err != nil {
+		b.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
+	}
+	msg, err := NewSentMessage(params)
+	if err != nil {
+		b.Fatalf("failed to create new message with seed %d: %s.", seed, err)
+	}
+	env, err := msg.Wrap(params)
+	if err != nil {
+		b.Fatalf("failed Wrap with seed %d: %s.", seed, err)
+	}
+
+	fs := generateFilters()
+
+	for i := 0; i < numOfFilters; i++ {
+		f, err := generateFilter(true)
+		if err != nil {
+			b.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+		}
+
+		_, err = fs.Install(f)
+		if err != nil {
+			b.Fatalf("failed install filter with seed %d: %s.", seed, err)
+		}
+
+	}
+
+	var topic TopicType
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		mrand.Read(topic[:])
+		env.Topic = topic
+
+		fs.NotifyWatchers(env, false)
 	}
 }

whisper/whisperv5/whisper_test.go

@@ -542,7 +542,7 @@ func TestCustomization(t *testing.T) {
 
 	const smallPoW = 0.00001
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -636,7 +636,7 @@ func TestSymmetricSendCycle(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter1, err := generateFilter(t, true)
+	filter1, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -725,7 +725,7 @@ func TestSymmetricSendWithoutAKey(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter, err := generateFilter(t, true)
+	filter, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -793,7 +793,7 @@ func TestSymmetricSendKeyMismatch(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter, err := generateFilter(t, true)
+	filter, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}

whisper/whisperv6/filter.go

@@ -26,6 +26,11 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const (
+	ALL_TOPICS        = ""
+	MAX_POOL_CAPACITY = 1000
+)
+
 // Filter represents a Whisper message filter
 type Filter struct {
 	Src        *ecdsa.PublicKey  // Sender of the message
@@ -42,16 +47,18 @@ type Filter struct {
 
 // Filters represents a collection of filters
 type Filters struct {
-	watchers map[string]*Filter
+	watchers     map[string]*Filter
-	whisper  *Whisper
+	whisper      *Whisper
-	mutex    sync.RWMutex
+	mutex        sync.RWMutex
+	topicMatcher *topicMatcher
 }
 
 // NewFilters returns a newly created filter collection
 func NewFilters(w *Whisper) *Filters {
 	return &Filters{
-		watchers: make(map[string]*Filter),
+		watchers:     make(map[string]*Filter),
-		whisper:  w,
+		whisper:      w,
+		topicMatcher: newTopicMatcher(),
 	}
 }
 
@@ -82,6 +89,8 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 	}
 
 	fs.watchers[id] = watcher
+	//add topic matching for watcher
+	fs.topicMatcher.addFilterToTopicsMapping(watcher, id)
 	return id, err
 }
 
@@ -92,6 +101,7 @@ func (fs *Filters) Uninstall(id string) bool {
 	defer fs.mutex.Unlock()
 	if fs.watchers[id] != nil {
 		delete(fs.watchers, id)
+		fs.topicMatcher.removeTopicFromTopicMapping(id)
 		return true
 	}
 	return false
@@ -108,15 +118,22 @@ func (fs *Filters) Get(id string) *Filter {
 // for the envelope's topic.
 func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	var msg *ReceivedMessage
+	matchedTopics := fs.topicMatcher.take()
+	defer fs.topicMatcher.resolve(matchedTopics)
 
 	fs.mutex.RLock()
 	defer fs.mutex.RUnlock()
 
-	i := -1 // only used for logging info
+	fs.topicMatcher.matchedTopics(env.Topic, &matchedTopics)
-	for _, watcher := range fs.watchers {
+	for _, watcherID := range matchedTopics {
-		i++
+		watcher, ok := fs.watchers[watcherID]
+		if !ok {
+			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: filter not exists", env.Hash(), watcherID))
+			continue
+		}
+
 		if p2pMessage && !watcher.AllowP2P {
-			log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), i))
+			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcherID))
 			continue
 		}
 
@@ -128,10 +145,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 			if match {
 				msg = env.Open(watcher)
 				if msg == nil {
-					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i)
+					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcherID)
 				}
 			} else {
-				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i)
+				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcherID)
 			}
 		}
 
@@ -201,9 +218,9 @@ func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
 	}
 
 	if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
-		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) && f.MatchTopic(msg.Topic)
+		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst)
 	} else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
-		return f.SymKeyHash == msg.SymKeyHash && f.MatchTopic(msg.Topic)
+		return f.SymKeyHash == msg.SymKeyHash
 	}
 	return false
 }
@@ -216,38 +233,6 @@ func (f *Filter) MatchEnvelope(envelope *Envelope) bool {
 		return false
 	}
 
-	return f.MatchTopic(envelope.Topic)
-}
-
-// MatchTopic checks that the filter captures a given topic.
-func (f *Filter) MatchTopic(topic TopicType) bool {
-	if len(f.Topics) == 0 {
-		// any topic matches
-		return true
-	}
-
-	for _, bt := range f.Topics {
-		if matchSingleTopic(topic, bt) {
-			return true
-		}
-	}
-	return false
-}
-
-func matchSingleTopic(topic TopicType, bt []byte) bool {
-	if len(bt) > TopicLength {
-		bt = bt[:TopicLength]
-	}
-
-	if len(bt) < TopicLength {
-		return false
-	}
-
-	for j, b := range bt {
-		if topic[j] != b {
-			return false
-		}
-	}
 	return true
 }
 
@@ -261,3 +246,91 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
 	// the curve is always the same, just compare the points
 	return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0
 }
+
+//topicMatcher keeps topic->watcher mapping
+type topicMatcher struct {
+	//structure - map[topic]map[filterID]
+	//mapping for topics
+	//"" topic means that the filter allows all topic values
+	mapper map[string]map[string]struct{}
+	mx     sync.RWMutex
+	pool   sync.Pool
+}
+
+//newTopicMatcher returns a newly created topic matcher
+func newTopicMatcher() *topicMatcher {
+	tm := new(topicMatcher)
+	tm.mapper = make(map[string]map[string]struct{})
+	tm.mapper[ALL_TOPICS] = make(map[string]struct{})
+	tm.pool.New = func() interface{} {
+		return []string{}
+	}
+	return tm
+}
+
+//take returns []string from pool
+func (fs *topicMatcher) take() []string {
+	return fs.pool.Get().([]string)
+}
+
+//resolve put []string to pool
+func (fs *topicMatcher) resolve(s []string) {
+	if cap(s) > MAX_POOL_CAPACITY {
+		return
+	}
+	fs.pool.Put(s[:0])
+}
+
+//addFilterToTopicsMapping fill topic->watcher mapping for current watcher
+func (fs *topicMatcher) addFilterToTopicsMapping(watcher *Filter, id string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+
+	for i := range fs.prepareTopicsMapping(watcher) {
+		topicMapping, ok := fs.mapper[i]
+		if !ok {
+			fs.mapper[i] = make(map[string]struct{})
+			topicMapping = fs.mapper[i]
+		}
+		topicMapping[id] = struct{}{}
+	}
+}
+
+//removeTopicFromTopicMapping removes mapping info by filterID
+func (fs *topicMatcher) removeTopicFromTopicMapping(filterID string) {
+	fs.mx.Lock()
+	defer fs.mx.Unlock()
+	for i := range fs.mapper {
+		delete(fs.mapper[i], filterID)
+	}
+}
+
+//prepareTopicsMapping returns set of topics for watcher
+func (fs *topicMatcher) prepareTopicsMapping(watcher *Filter) map[string]struct{} {
+	topics := make(map[string]struct{}, len(watcher.Topics))
+
+	if len(watcher.Topics) == 0 {
+		topics[ALL_TOPICS] = struct{}{}
+		return topics
+	}
+
+	for _, topic := range watcher.Topics {
+		topics[common.ToHex(topic)] = struct{}{}
+	}
+
+	return topics
+}
+
+//matchedTopics write all matched topics to matched
+func (fs *topicMatcher) matchedTopics(topic TopicType, matched *[]string) {
+	fs.mx.RLock()
+	defer fs.mx.RUnlock()
+
+	for i := range fs.mapper[ALL_TOPICS] {
+		*matched = append(*matched, i)
+	}
+
+	for i := range fs.mapper[topic.String()] {
+		*matched = append(*matched, i)
+	}
+}

whisper/whisperv6/filter_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"fmt"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 )
@@ -48,7 +49,7 @@ type FilterTestCase struct {
 	msgCnt int
 }
 
-func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
+func generateFilter(symmetric bool) (*Filter, error) {
 	var f Filter
 	f.Messages = make(map[common.Hash]*ReceivedMessage)
 
@@ -62,8 +63,7 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 
 	key, err := crypto.GenerateKey()
 	if err != nil {
-		t.Fatalf("generateFilter 1 failed with seed %d.", seed)
+		return nil, fmt.Errorf("generateFilter 1 failed with seed %d. Error: %s", seed, err.Error())
-		return nil, err
 	}
 	f.Src = &key.PublicKey
 
@@ -74,8 +74,7 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 	} else {
 		f.KeyAsym, err = crypto.GenerateKey()
 		if err != nil {
-			t.Fatalf("generateFilter 2 failed with seed %d.", seed)
+			return nil, fmt.Errorf("generateFilter 2 failed with seed %d. Error: %s", seed, err.Error())
-			return nil, err
 		}
 	}
 
@@ -83,10 +82,21 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) {
 	return &f, nil
 }
 
+func generateFilters() *Filters {
+	fs := Filters{
+		watchers:     make(map[string]*Filter),
+		topicMatcher: newTopicMatcher(),
+	}
+	return &fs
+}
+
 func generateTestCases(t *testing.T, SizeTestFilters int) []FilterTestCase {
 	cases := make([]FilterTestCase, SizeTestFilters)
 	for i := 0; i < SizeTestFilters; i++ {
-		f, _ := generateFilter(t, true)
+		f, err := generateFilter(true)
+		if err != nil {
+			t.Fatal(err)
+		}
 		cases[i].f = f
 		cases[i].alive = mrand.Int()&int(1) == 0
 	}
@@ -137,8 +147,10 @@ func TestInstallSymKeyGeneratesHash(t *testing.T) {
 
 	w := New(&Config{})
 	filters := NewFilters(w)
-	filter, _ := generateFilter(t, true)
+	filter, err := generateFilter(true)
-
+	if err != nil {
+		t.Fatal(err)
+	}
 	// save the current SymKeyHash for comparison
 	initialSymKeyHash := filter.SymKeyHash
 
@@ -146,7 +158,7 @@ func TestInstallSymKeyGeneratesHash(t *testing.T) {
 	var invalid common.Hash
 	filter.SymKeyHash = invalid
 
-	_, err := filters.Install(filter)
+	_, err = filters.Install(filter)
 
 	if err != nil {
 		t.Fatalf("Error installing the filter: %s", err)
@@ -164,7 +176,10 @@ func TestInstallIdenticalFilters(t *testing.T) {
 
 	w := New(&Config{})
 	filters := NewFilters(w)
-	filter1, _ := generateFilter(t, true)
+	filter1, err := generateFilter(true)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// Copy the first filter since some of its fields
 	// are randomly gnerated.
@@ -176,7 +191,7 @@ func TestInstallIdenticalFilters(t *testing.T) {
 		Messages: make(map[common.Hash]*ReceivedMessage),
 	}
 
-	_, err := filters.Install(filter1)
+	_, err = filters.Install(filter1)
 
 	if err != nil {
 		t.Fatalf("Error installing the first filter with seed %d: %s", seed, err)
@@ -234,7 +249,10 @@ func TestInstallFilterWithSymAndAsymKeys(t *testing.T) {
 
 	w := New(&Config{})
 	filters := NewFilters(w)
-	filter1, _ := generateFilter(t, true)
+	filter1, err := generateFilter(true)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	asymKey, err := crypto.GenerateKey()
 	if err != nil {
@@ -288,12 +306,12 @@ func TestComparePubKey(t *testing.T) {
 func TestMatchEnvelope(t *testing.T) {
 	InitSingleTest()
 
-	fsym, err := generateFilter(t, true)
+	fsym, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
 
-	fasym, err := generateFilter(t, false)
+	fasym, err := generateFilter(false)
 	if err != nil {
 		t.Fatalf("failed generateFilter() with seed %d: %s.", seed, err)
 	}
@@ -314,19 +332,8 @@ func TestMatchEnvelope(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
 	}
-	match := fsym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope symmetric with seed %d.", seed)
-	}
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope asymmetric with seed %d.", seed)
-	}
 
 	// encrypt symmetrically
-	i := mrand.Int() % 4
-	fsym.Topics[i] = params.Topic[:]
-	fasym.Topics[i] = params.Topic[:]
 	msg, err = NewSentMessage(params)
 	if err != nil {
 		t.Fatalf("failed to create new message with seed %d: %s.", seed, err)
@@ -336,20 +343,20 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed Wrap() with seed %d: %s.", seed, err)
 	}
 
-	// symmetric + matching topic: match
+	// symmetric
-	match = fsym.MatchEnvelope(env)
+	match := fsym.MatchEnvelope(env)
 	if !match {
 		t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed)
 	}
 
-	// symmetric + matching topic + insufficient PoW: mismatch
+	// symmetric + insufficient PoW: mismatch
 	fsym.PoW = env.PoW() + 1.0
 	match = fsym.MatchEnvelope(env)
 	if match {
 		t.Fatalf("failed MatchEnvelope(symmetric + matching topic + insufficient PoW) asymmetric with seed %d.", seed)
 	}
 
-	// symmetric + matching topic + sufficient PoW: match
+	// symmetric + sufficient PoW: match
 	fsym.PoW = env.PoW() / 2
 	match = fsym.MatchEnvelope(env)
 	if !match {
@@ -387,26 +394,6 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(encryption method mismatch) with seed %d.", seed)
 	}
 
-	// asymmetric + mismatching topic: mismatch
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + mismatching topic) with seed %d.", seed)
-	}
-
-	// asymmetric + matching topic: match
-	fasym.Topics[i] = fasym.Topics[i+1]
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + matching topic) with seed %d.", seed)
-	}
-
-	// asymmetric + filter without topic (wildcard): match
-	fasym.Topics = nil
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(asymmetric + filter without topic) with seed %d.", seed)
-	}
-
 	// asymmetric + insufficient PoW: mismatch
 	fasym.PoW = env.PoW() + 1.0
 	match = fasym.MatchEnvelope(env)
@@ -420,20 +407,6 @@ func TestMatchEnvelope(t *testing.T) {
 	if !match {
 		t.Fatalf("failed MatchEnvelope(asymmetric + sufficient PoW) with seed %d.", seed)
 	}
-
-	// filter without topic + envelope without topic: match
-	env.Topic = TopicType{}
-	match = fasym.MatchEnvelope(env)
-	if !match {
-		t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed)
-	}
-
-	// filter with topic + envelope without topic: mismatch
-	fasym.Topics = fsym.Topics
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed)
-	}
 }
 
 func TestMatchMessageSym(t *testing.T) {
@@ -444,7 +417,7 @@ func TestMatchMessageSym(t *testing.T) {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
@@ -485,13 +458,6 @@ func TestMatchMessageSym(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed)
 	}
 
-	// topic mismatch
-	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
-		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
-	}
-	f.Topics[index][0]--
-
 	// key mismatch
 	f.SymKeyHash[0]++
 	if f.MatchMessage(msg) {
@@ -530,7 +496,7 @@ func TestMatchMessageSym(t *testing.T) {
 func TestMatchMessageAsym(t *testing.T) {
 	InitSingleTest()
 
-	f, err := generateFilter(t, false)
+	f, err := generateFilter(false)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
@@ -578,13 +544,6 @@ func TestMatchMessageAsym(t *testing.T) {
 		t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed)
 	}
 
-	// topic mismatch
-	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
-		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
-	}
-	f.Topics[index][0]--
-
 	// key mismatch
 	prev := *f.KeyAsym.PublicKey.X
 	zero := *big.NewInt(0)
@@ -815,58 +774,215 @@ func TestVariableTopics(t *testing.T) {
 		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
 	}
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
 	}
 
+	fs := generateFilters()
+
+	filterID, err := fs.Install(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	for i := 0; i < 4; i++ {
 		env.Topic = BytesToTopic(f.Topics[i])
+		//test match
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
 		match = f.MatchEnvelope(env)
-		if !match {
+		if !(match && hasFilterID(matched, filterID)) {
 			t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
 
-		f.Topics[i][lastTopicByte]++
+		//change envelop topic
+		env.Topic[lastTopicByte]++
+
+		//false positive test
 		match = f.MatchEnvelope(env)
-		if match {
+
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if match && hasFilterID(matched, filterID) {
 			t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i)
 		}
 	}
 }
 
-func TestMatchSingleTopic_ReturnTrue(t *testing.T) {
+func TestTopicsMapping(t *testing.T) {
-	bt := []byte("test")
+	InitSingleTest()
-	topic := BytesToTopic(bt)
 
-	if !matchSingleTopic(topic, bt) {
+	const lastTopicByte = 3
-		t.FailNow()
+	params, err := generateMessageParams()
+	if err != nil {
+		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
+	}
+	msg, err := NewSentMessage(params)
+	if err != nil {
+		t.Fatalf("failed to create new message with seed %d: %s.", seed, err)
+	}
+	env, err := msg.Wrap(params)
+	if err != nil {
+		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
+	}
+
+	f, err := generateFilter(true)
+	if err != nil {
+		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+	}
+	fs := generateFilters()
+
+	for i := 0; i < len(f.Topics); i++ {
+		env.Topic = BytesToTopic(f.Topics[i])
+
+		//test match
+		filterID, err := fs.Install(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		matched := []string{}
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if !hasFilterID(matched, filterID) {
+			t.Fatalf("failed matchedTopics, step %d.", i)
+		}
+
+		//test match without filter
+		if !fs.Uninstall(filterID) {
+			t.Fatal("Failed to uninstall filter")
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			t.Fatalf("failed match without filter, step %d.", i)
+		}
+
+		//test match with changed topic
+		f.Topics[i][lastTopicByte]++
+		filterID, err = fs.Install(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		matched = matched[:0]
+		fs.topicMatcher.matchedTopics(env.Topic, &matched)
+		if hasFilterID(matched, filterID) {
+			t.Fatalf("failed match with changed topic, step %d.", i)
+		}
+		if !fs.Uninstall(filterID) {
+			t.Fatal("Failed to uninstall filter")
+		}
 	}
 }
 
-func TestMatchSingleTopic_WithTail_ReturnTrue(t *testing.T) {
+func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) {
-	bt := []byte("test with tail")
+	InitSingleTest()
-	topic := BytesToTopic([]byte("test"))
 
-	if !matchSingleTopic(topic, bt) {
+	f, err := generateFilter(true)
-		t.FailNow()
+	if err != nil {
+		t.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+	}
+	fs := generateFilters()
+
+	f.Topics = [][]byte{}
+
+	filterID, err := fs.Install(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	//generate topic
+	topic := TopicType{}
+	mrand.Read(topic[:])
+
+	matched := []string{}
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if !hasFilterID(matched, filterID) {
+		t.Fatal("failed matchedTopics")
+	}
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; !ok {
+		t.Fatal("watcher mapping incorrect")
+	}
+
+	//test match without filter
+	if !fs.Uninstall(filterID) {
+		t.Fatal("Failed to uninstall filter")
+	}
+	matched = matched[:0]
+	fs.topicMatcher.matchedTopics(topic, &matched)
+	if hasFilterID(matched, filterID) {
+		t.Fatal("failed match without filter")
+	}
+
+	if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; ok {
+		t.Fatal("watcher mapping incorrect")
 	}
 }
 
-func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) {
+func hasFilterID(matched []string, filterID string) bool {
-	bt := []byte("tes")
+	for i := range matched {
-	topic := BytesToTopic(bt)
+		if matched[i] == filterID {
-
+			return true
-	if matchSingleTopic(topic, bt) {
+		}
-		t.FailNow()
 	}
+	return false
 }
 
-func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) {
+func BenchmarkFilter_MatchEnvelope_5Filters(b *testing.B) {
-	bt := []byte("test")
+	InitSingleTest()
-	topic := BytesToTopic([]byte("not_equal"))
+	benchFilter_MatchMessage(b, 5)
+}
+func BenchmarkFilter_MatchEnvelope_10Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 10)
+}
+func BenchmarkFilter_MatchEnvelope_20Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 20)
+}
+func BenchmarkFilter_MatchEnvelope_50Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 50)
+}
+func BenchmarkFilter_MatchEnvelope_100Filters(b *testing.B) {
+	InitSingleTest()
+	benchFilter_MatchMessage(b, 100)
+}
 
-	if matchSingleTopic(topic, bt) {
+func benchFilter_MatchMessage(b *testing.B, numOfFilters int) {
-		t.FailNow()
+	params, err := generateMessageParams()
+	if err != nil {
+		b.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
+	}
+	msg, err := NewSentMessage(params)
+	if err != nil {
+		b.Fatalf("failed to create new message with seed %d: %s.", seed, err)
+	}
+	env, err := msg.Wrap(params)
+	if err != nil {
+		b.Fatalf("failed Wrap with seed %d: %s.", seed, err)
+	}
+
+	fs := generateFilters()
+
+	for i := 0; i < numOfFilters; i++ {
+		f, err := generateFilter(true)
+		if err != nil {
+			b.Fatalf("failed generateFilter with seed %d: %s.", seed, err)
+		}
+
+		_, err = fs.Install(f)
+		if err != nil {
+			b.Fatalf("failed install filter with seed %d: %s.", seed, err)
+		}
+
+	}
+
+	var topic TopicType
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		mrand.Read(topic[:])
+		env.Topic = topic
+
+		fs.NotifyWatchers(env, false)
 	}
 }

whisper/whisperv6/whisper_test.go

@@ -524,7 +524,7 @@ func TestCustomization(t *testing.T) {
 
 	const smallPoW = 0.00001
 
-	f, err := generateFilter(t, true)
+	f, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -618,7 +618,7 @@ func TestSymmetricSendCycle(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter1, err := generateFilter(t, true)
+	filter1, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -707,7 +707,7 @@ func TestSymmetricSendWithoutAKey(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter, err := generateFilter(t, true)
+	filter, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
@@ -775,7 +775,7 @@ func TestSymmetricSendKeyMismatch(t *testing.T) {
 	w.Start(nil)
 	defer w.Stop()
 
-	filter, err := generateFilter(t, true)
+	filter, err := generateFilter(true)
 	if err != nil {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}