Skip to content

Commit

Permalink
Improve: delete useless code and code coverage is now 100%
Browse files Browse the repository at this point in the history
  • Loading branch information
Dreamacro committed Jul 12, 2018
1 parent 283e4e1 commit 39b4551
Show file tree
Hide file tree
Showing 4 changed files with 22 additions and 46 deletions.
15 changes: 0 additions & 15 deletions observable/iterable.go
Original file line number Diff line number Diff line change
@@ -1,18 +1,3 @@
package observable

import (
"errors"
)

type Iterable <-chan interface{}

func NewIterable(any interface{}) (Iterable, error) {
switch any := any.(type) {
case chan interface{}:
return Iterable(any), nil
case <-chan interface{}:
return Iterable(any), nil
default:
return nil, errors.New("type error")
}
}
1 change: 0 additions & 1 deletion observable/observable.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@ func (o *Observable) Subscribe() (Subscription, error) {
func (o *Observable) UnSubscribe(sub Subscription) {
elm, exist := o.listener.Load(sub)
if !exist {
println("not exist")
return
}
subscriber := elm.(*Subscriber)
Expand Down
37 changes: 22 additions & 15 deletions observable/observable_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,7 @@ func TestObservable(t *testing.T) {
t.Error(err)
}
count := 0
for {
_, open := <-data
if !open {
break
}
for range data {
count = count + 1
}
if count != 5 {
Expand All @@ -49,11 +45,7 @@ func TestObservable_MutilSubscribe(t *testing.T) {
var wg sync.WaitGroup
wg.Add(2)
waitCh := func(ch <-chan interface{}) {
for {
_, open := <-ch
if !open {
break
}
for range ch {
count = count + 1
}
wg.Done()
Expand All @@ -80,6 +72,25 @@ func TestObservable_UnSubscribe(t *testing.T) {
}
}

func TestObservable_SubscribeClosedSource(t *testing.T) {
iter := iterator([]interface{}{1})
src := NewObservable(iter)
data, _ := src.Subscribe()
<-data

_, closed := src.Subscribe()
if closed == nil {
t.Error("Observable should be closed")
}
}

func TestObservable_UnSubscribeWithNotExistSubscription(t *testing.T) {
sub := Subscription(make(chan interface{}))
iter := iterator([]interface{}{1})
src := NewObservable(iter)
src.UnSubscribe(sub)
}

func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
// waiting for other goroutine recycle
time.Sleep(120 * time.Millisecond)
Expand All @@ -97,11 +108,7 @@ func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
var wg sync.WaitGroup
wg.Add(max)
waitCh := func(ch <-chan interface{}) {
for {
_, open := <-ch
if !open {
break
}
for range ch {
}
wg.Done()
}
Expand Down
15 changes: 0 additions & 15 deletions observable/util.go

This file was deleted.

0 comments on commit 39b4551

Please sign in to comment.