internal/dispatcher/dispatcher.go

@@ -113,21 +113,12 @@ func (d *Dispatcher) dispatchMessage(msg *message.Message) {
 				return
 			}
 			log.Println("Websocket terminated: reconnecting")
-			err := d.Disconnect()
-			if err != nil {
-				log.Println(err)
-			}
 			d.Connect()
-			if err != nil {
-				log.Println(err)
-			}
-
 			subsList := d.store.GetAll()
 			for i := range subsList {
 				sub := subsList[i]
 				d.Subscribe(sub.Name(), sub.AuthToken())
 			}
-
 		case message.MetaSubscribe:
 			//handle MetaSubscribe resp
 			d.pendingSubsMu.Lock()

subscription/subscription.go

@@ -32,18 +32,11 @@ func NewSubscription(chanel string, unsub Unsubscriber, authToken string, msgCh
 
 func (s *Subscription) OnMessage(onMessage func(channel string, msg message.Data)) error {
 	var inMsg *message.Message
-	go s.StartMessageLoop(inMsg, onMessage)
-
-	return nil
-}
-
-func (s *Subscription) StartMessageLoop(inMsg *message.Message, callback func(channel string, msg message.Data)) error {
 	for inMsg = range s.msgCh {
 		if inMsg.GetError() != nil {
-
 			return inMsg.GetError()
 		}
-		callback(inMsg.Channel, inMsg.Data)
+		onMessage(inMsg.Channel, inMsg.Data)
 	}
 	return nil
 }

transport/websocket/websocket.go

@@ -46,10 +46,6 @@ func (w *Websocket) Init(endpoint string, options *transport.Options) error {
 
 	w.stopCh = make(chan error)
 	w.conn, _, err = websocket.DefaultDialer.Dial(endpoint, options.Headers)
-	if err != nil {
-		return err
-	}
-
 	err = w.conn.UnderlyingConn().(*tls.Conn).NetConn().(*net.TCPConn).SetKeepAlive(true)
 	if err != nil {
 		return err
@@ -142,8 +138,7 @@ func (w *Websocket) Handshake(msg *message.Message) (resp *message.Message, err
 // a connection is established by sending a message to the /meta/connect channel
 func (w *Websocket) Connect(msg *message.Message) error {
 	go func() {
-		err := w.readWorker()
+		log.Fatal(w.readWorker())
-		log.Fatal(err)
 	}()
 	return w.SendMessage(msg)
 }