fix process kill

fix unsubscribe error
This commit is contained in:
Marcelo Pires 2018-09-06 18:40:00 +02:00
parent db6b6c3848
commit e98ccfdec8
2 changed files with 26 additions and 24 deletions

View File

@ -1,7 +1,7 @@
package faye_test package faye_test
import ( import (
"fmt" "github.com/pkg/errors"
"github.com/thesyncim/faye" "github.com/thesyncim/faye"
"github.com/thesyncim/faye/extensions" "github.com/thesyncim/faye/extensions"
"github.com/thesyncim/faye/message" "github.com/thesyncim/faye/message"
@ -15,6 +15,8 @@ import (
type cancelFn func() type cancelFn func()
var unauthorizedErr = errors.New("500::unauthorized channel")
func setup(t *testing.T) (cancelFn, error) { func setup(t *testing.T) (cancelFn, error) {
cmd := exec.Command("npm", "start") cmd := exec.Command("npm", "start")
cmd.Stdout = os.Stdout cmd.Stdout = os.Stdout
@ -26,14 +28,14 @@ func setup(t *testing.T) (cancelFn, error) {
} }
var cancel = func() { var cancel = func() {
exec.Command("taskkill", "/F", "/T", "/PID", fmt.Sprint(cmd.Process.Pid)).Run() cmd.Process.Signal(os.Kill)
log.Println("canceled") log.Println("canceled")
} }
go func() { go func() {
select { select {
case <-time.After(time.Second * 30): case <-time.After(time.Second * 30):
cancel() cancel()
t.Fatal("failed") t.Fatal("test failed")
os.Exit(1) os.Exit(1)
} }
@ -42,7 +44,7 @@ func setup(t *testing.T) (cancelFn, error) {
return cancel, nil return cancel, nil
} }
func TestServerSubscribeAndPublish(t *testing.T) { func TestServerSubscribeAndPublish10Messages(t *testing.T) {
shutdown, err := setup(t) shutdown, err := setup(t)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@ -56,21 +58,22 @@ func TestServerSubscribeAndPublish(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
var delivered int
var done sync.WaitGroup
done.Add(10)
client.OnPublishResponse("/test", func(message *message.Message) { client.OnPublishResponse("/test", func(message *message.Message) {
if !message.Successful { if !message.Successful {
t.Fatalf("failed to send message with id %s", message.Id) t.Fatalf("failed to send message with id %s", message.Id)
} }
delivered++
done.Done()
}) })
var done sync.WaitGroup
done.Add(10)
var delivered int
go func() { go func() {
client.Subscribe("/test", func(data message.Data) { client.Subscribe("/test", func(data message.Data) {
if data != "hello world" { if data != "hello world" {
t.Fatalf("expecting: `hello world` got : %s", data) t.Fatalf("expecting: `hello world` got : %s", data)
} }
delivered++
done.Done()
}) })
}() }()
@ -89,11 +92,15 @@ func TestServerSubscribeAndPublish(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
//try to publish one more message
id, err := client.Publish("/test", "hello world")
if err != nil {
t.Fatal(err)
}
log.Println(id)
if delivered != 10 { if delivered != 10 {
t.Fatal("message received after client unsubscribe") t.Fatal("message received after client unsubscribe")
} }
log.Println("complete")
} }
func TestSubscribeUnauthorizedChannel(t *testing.T) { func TestSubscribeUnauthorizedChannel(t *testing.T) {
@ -113,8 +120,8 @@ func TestSubscribeUnauthorizedChannel(t *testing.T) {
err = client.Subscribe("/unauthorized", func(data message.Data) { err = client.Subscribe("/unauthorized", func(data message.Data) {
t.Fatal("received message on unauthorized channel") t.Fatal("received message on unauthorized channel")
}) })
if err == nil { if err.Error() != unauthorizedErr.Error() {
t.Fatal("subscribed to an unauthorized channel") t.Fatalf("expecting `500::unauthorized channel` got : `%s`", err.Error())
} }
} }

View File

@ -89,8 +89,8 @@ func (w *Websocket) readWorker() error {
if !ok { if !ok {
panic("BUG: subscription not registered `" + msg.Subscription + "`") panic("BUG: subscription not registered `" + msg.Subscription + "`")
} }
if msg.GetError() != nil { if msg.GetError() == nil {
//inject the error //inject the error if the server returns unsuccessful without error
msg.Error = fmt.Sprintf("susbscription `%s` failed", msg.Subscription) msg.Error = fmt.Sprintf("susbscription `%s` failed", msg.Subscription)
} }
subscription <- msg subscription <- msg
@ -99,16 +99,7 @@ func (w *Websocket) readWorker() error {
delete(w.subs, msg.Channel) delete(w.subs, msg.Channel)
w.subsMu.Unlock() w.subsMu.Unlock()
} }
case transport.MetaUnsubscribe:
//handle MetaUnsubscribe resp
case transport.MetaConnect:
//handle MetaConnect resp
case transport.MetaDisconnect:
//handle MetaDisconnect resp
case transport.MetaHandshake:
//handle MetaHandshake resp
} }
continue continue
@ -277,6 +268,10 @@ func (w *Websocket) Unsubscribe(subscription string) error {
delete(w.subs, subscription) delete(w.subs, subscription)
} }
w.subsMu.Unlock() w.subsMu.Unlock()
//remove onPublishResponse handler
w.onPubResponseMu.Lock()
delete(w.onPublishResponse, subscription)
w.onPubResponseMu.Unlock()
return w.sendMessage(m) return w.sendMessage(m)
} }