add getStream extension

This commit is contained in:
Marcelo Pires 2018-09-05 10:56:45 +02:00
parent 3c187e364b
commit d9385e1506
2 changed files with 38 additions and 7 deletions

27
extensions/getstream.go Normal file
View File

@ -0,0 +1,27 @@
package extensions
import (
"github.com/thesyncim/faye/message"
"github.com/thesyncim/faye/transport"
)
type GetStream struct {
UserID string `json:"user_id,omitempty"`
ApiKey string `json:"api_key,omitempty"`
Signature string `json:"signature,omitempty"`
}
func NewGetStream(apiKey string, signature string) GetStream {
return GetStream{
ApiKey: apiKey,
Signature: signature,
}
}
func (gt GetStream) OutExtension(message *message.Message) {
if message.Channel == string(transport.Subscribe) {
//get useriID
gt.UserID = message.Subscription[1:]
message.Ext = gt
}
}

View File

@ -49,7 +49,8 @@ func (w *Websocket) Options() *transport.Options {
return w.TransportOpts return w.TransportOpts
} }
func (w *Websocket) Handshake() (err error) { func (w *Websocket) Handshake() (err error) {
if err = w.conn.WriteJSON(append([]message.Message{}, message.Message{ var payload []message.Message
if err = w.conn.WriteJSON(append(payload, message.Message{
Channel: string(transport.Handshake), Channel: string(transport.Handshake),
Version: "1.0", //todo const Version: "1.0", //todo const
SupportedConnectionTypes: []string{transportName}, SupportedConnectionTypes: []string{transportName},
@ -71,8 +72,9 @@ func (w *Websocket) Handshake() (err error) {
} }
func (w *Websocket) Connect() error { func (w *Websocket) Connect() error {
var payload []message.Message
//todo verify if extensions are applied on connect,verify if hs is complete //todo verify if extensions are applied on connect,verify if hs is complete
return w.conn.WriteJSON(append([]message.Message{}, message.Message{ return w.conn.WriteJSON(append(payload, message.Message{
Channel: string(transport.Connect), Channel: string(transport.Connect),
ClientId: w.clientID, ClientId: w.clientID,
ConnectionType: transportName, ConnectionType: transportName,
@ -90,7 +92,9 @@ func (w *Websocket) Subscribe(subscription string, onMessage func(message *messa
if w.TransportOpts.OutExt != nil { if w.TransportOpts.OutExt != nil {
w.TransportOpts.OutExt(m) w.TransportOpts.OutExt(m)
} }
err := w.conn.WriteJSON(append([]message.Message{}, *m))
var payload []message.Message
err := w.conn.WriteJSON(append(payload, *m))
if err != nil { if err != nil {
return err return err
} }
@ -108,7 +112,7 @@ func (w *Websocket) Subscribe(subscription string, onMessage func(message *messa
//report err just for sanity //report err just for sanity
} }
unsubsCh := make(chan struct{}, 0) unsubsCh := make(chan struct{}, 0)
//todo multiple subs
w.subs[subscription] = unsubsCh w.subs[subscription] = unsubsCh
for { for {
@ -118,13 +122,13 @@ func (w *Websocket) Subscribe(subscription string, onMessage func(message *messa
default: default:
} }
//todo guard unsusribe //todo guard unsusribe
var hsResps []message.Message var payload []message.Message
err := w.conn.ReadJSON(&hsResps) err := w.conn.ReadJSON(&payload)
if err != nil { if err != nil {
return err return err
} }
msg := hsResps[0] msg := payload[0]
onMessage(&msg) onMessage(&msg)
} }
return nil return nil