Implement filtering

pfs.go

@@ -24,7 +24,7 @@ func New() *PFS {
 	}
 }
 
-func (p *PFS) Pub(args ...interface{}) {
+func (p *PFS) Pub(args ...interface{}) bool {
 	p.lmu.Lock()
 	defer p.lmu.Unlock()
 
@@ -33,8 +33,11 @@ func (p *PFS) Pub(args ...interface{}) {
 		arguments = append(arguments, reflect.ValueOf(v))
 	}
 
-	wg := sync.WaitGroup{}
+	if !p.filtering(arguments) {
+		return false
+	}
 
+	wg := sync.WaitGroup{}
 	wg.Add(len(p.listeners))
 	for _, fn := range p.listeners {
 		go func(f reflect.Value) {
@@ -44,6 +47,7 @@ func (p *PFS) Pub(args ...interface{}) {
 	}
 
 	wg.Wait()
+	return true
 }
 
 func (p *PFS) Sub(f interface{}) error {
@@ -99,3 +103,17 @@ func (p *PFS) Filter(f interface{}) error {
 
 	return nil
 }
+
+func (p *PFS) filtering(arguments []reflect.Value) bool {
+	p.fmu.RLock()
+	defer p.fmu.RUnlock()
+
+	for _, fn := range p.filters {
+		res := fn.Call(arguments)
+		ok := res[0].Bool()
+		if !ok {
+			return false
+		}
+	}
+	return true
+}

pfs_test.go

@@ -49,19 +49,25 @@ func TestFilter(t *testing.T) {
 	}
 
 	err = p.Filter(func(j int) bool {
-		return j == 3
+		return j != 3
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	p.Pub(2)
+	ok := p.Pub(2)
 	if i != 3 {
 		t.Errorf("Expected i == 3, Got i == %d", i)
 	}
+	if !ok {
+		t.Error("should return true When not reject. But got false.")
+	}
 
-	p.Pub(3)
+	ok = p.Pub(3)
 	if i != 3 {
 		t.Errorf("Expected i == 3, Got i == %d", i)
 	}
+	if ok {
+		t.Error("should return false When reject. But got true.")
+	}
 }