More refactoring; eventstream is working, but still needs work.

This commit is contained in:
Jeff Walter 2018-09-19 02:07:32 -05:00
parent 011447ef60
commit 930ecd3bad
12 changed files with 61 additions and 62 deletions

4
Gopkg.lock generated
View File

@ -11,11 +11,11 @@
[[projects]] [[projects]]
branch = "master" branch = "master"
digest = "1:63eb8e7863bcc41b162d76c3e1b5dd4991fd01f29585017b3a8bab7a5928edd3" digest = "1:b44621ef75eb48a8c09e7d6ce4d3f6681534307189c9bc0267c9532094c20a2a"
name = "github.com/r3labs/sse" name = "github.com/r3labs/sse"
packages = ["."] packages = ["."]
pruneopts = "" pruneopts = ""
revision = "ad82e5b42970ce737f0ce61490a1607331299496" revision = "507597500aa545836a85909d89a59e50d260f366"
[[projects]] [[projects]]
branch = "master" branch = "master"

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
import ( import (
"fmt" "fmt"

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
import ( import (
"github.com/jeffreydwalter/arlo-golang/internal/request" "github.com/jeffreydwalter/arlo-golang/internal/request"

View File

@ -1 +1 @@
package arlo_golang package arlo

View File

@ -1,9 +1,8 @@
package arlo_golang package arlo
import ( import (
"fmt" "fmt"
"github.com/jeffreydwalter/arlo-golang/internal/util"
"github.com/pkg/errors" "github.com/pkg/errors"
) )
@ -43,7 +42,7 @@ type Basestation struct {
type Basestations []Basestation type Basestations []Basestation
func (b *Basestation) Subscribe() (*Status, error) { func (b *Basestation) Subscribe() (*Status, error) {
b.eventStream = NewEventStream(BaseUrl+fmt.Sprintf(SubscribeUri, b.arlo.Account.Token), b.arlo.client.HttpClient, util.HeaderToMap(*b.arlo.client.BaseHttpHeader)) b.eventStream = NewEventStream(BaseUrl+fmt.Sprintf(SubscribeUri, b.arlo.Account.Token), b.arlo.client.HttpClient)
b.eventStream.Listen() b.eventStream.Listen()
transId := GenTransId() transId := GenTransId()
@ -71,22 +70,6 @@ func (b *Basestation) Subscribe() (*Status, error) {
return &status, nil return &status, nil
} }
/*
This is an example of the json you would pass in the body to UpdateFriends():
{
"firstName":"Some",
"lastName":"Body",
"devices":{
"XXXXXXXXXXXXX":"Camera 1",
"XXXXXXXXXXXXX":"Camera 2 ",
"XXXXXXXXXXXXX":"Camera 3"
},
"lastModified":1463977440911,
"adminUser":true,
"email":"user@example.com",
"id":"XXX-XXXXXXX"
}
*/
func (b *Basestation) GetState() (*NotifyResponse, error) { func (b *Basestation) GetState() (*NotifyResponse, error) {
transId := GenTransId() transId := GenTransId()
@ -100,11 +83,7 @@ func (b *Basestation) GetState() (*NotifyResponse, error) {
To: b.DeviceId, To: b.DeviceId,
} }
//fmt.Printf("BODY: %+v\n", body) b.eventStream.Subscriptions[transId] = make(chan *NotifyResponse)
//fmt.Printf("HEADERS: %+v\n", a.client.BaseHttpHeader)
fmt.Println("Subscribing to the eventstream.")
b.eventStream.Subscriptions[transId] = new(Subscriber)
resp, err := b.arlo.client.Post(fmt.Sprintf(NotifyUri, b.DeviceId), body, nil) resp, err := b.arlo.client.Post(fmt.Sprintf(NotifyUri, b.DeviceId), body, nil)
if err != nil { if err != nil {
@ -120,6 +99,37 @@ func (b *Basestation) GetState() (*NotifyResponse, error) {
return nil, errors.New("failed to get basestation status") return nil, errors.New("failed to get basestation status")
} }
notifyResponse := <-*b.eventStream.Subscriptions[transId] return <-b.eventStream.Subscriptions[transId], nil
return &notifyResponse, nil }
func (b *Basestation) GetCameraState() (*NotifyResponse, error) {
transId := GenTransId()
body := NotifyPayload{
Action: "get",
Resource: "cameras",
PublishResponse: false,
TransId: transId,
From: fmt.Sprintf("%s_%s", b.UserId, TransIdPrefix),
To: b.DeviceId,
}
b.eventStream.Subscriptions[transId] = make(chan *NotifyResponse)
resp, err := b.arlo.client.Post(fmt.Sprintf(NotifyUri, b.DeviceId), body, nil)
if err != nil {
return nil, errors.WithMessage(err, "failed to get basestation state")
}
var status Status
if err := resp.Decode(&status); err != nil {
return nil, err
}
if !status.Success {
return nil, errors.New("failed to get basestation status")
}
return <-b.eventStream.Subscriptions[transId], nil
} }

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
// A Camera is a Device of type "camera". // A Camera is a Device of type "camera".
// This type is here just for semantics. Some methods explicitly require a device of a certain type. // This type is here just for semantics. Some methods explicitly require a device of a certain type.

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
const ( const (
TransIdPrefix = "web" TransIdPrefix = "web"

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
import ( import (
"fmt" "fmt"

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
import ( import (
"bytes" "bytes"
@ -18,31 +18,27 @@ var (
FAILED_TO_SUBSCRIBE = errors.New("Failed to subscribe to SSEClient") FAILED_TO_SUBSCRIBE = errors.New("Failed to subscribe to SSEClient")
) )
type Subscriber chan NotifyResponse
type EventStream struct { type EventStream struct {
SSEClient *sse.Client
Subscriptions map[string]chan *NotifyResponse
Events chan *sse.Event
ErrorChan chan error
Registered bool Registered bool
Connected bool Connected bool
SSEClient *sse.Client
Events chan *sse.Event
Subscriptions map[string]*Subscriber
ErrorChan chan error
Responses map[string]NotifyResponse
Verbose bool Verbose bool
sync.Mutex sync.Mutex
} }
func NewEventStream(url string, client *http.Client, headers map[string]string) *EventStream { func NewEventStream(url string, client *http.Client) *EventStream {
SSEClient := sse.NewClient(url) SSEClient := sse.NewClient(url)
SSEClient.Connection = client SSEClient.Connection = client
SSEClient.Headers = headers
return &EventStream{ return &EventStream{
SSEClient: SSEClient, SSEClient: SSEClient,
Events: make(chan *sse.Event), Events: make(chan *sse.Event),
Subscriptions: map[string]*Subscriber{}, Subscriptions: make(map[string]chan *NotifyResponse),
ErrorChan: make(chan error, 1), ErrorChan: make(chan error, 1),
} }
} }
@ -59,9 +55,11 @@ func (e *EventStream) Listen() {
go func() { go func() {
for event := range e.Events { for event := range e.Events {
/*
fmt.Println("Got event message.") fmt.Println("Got event message.")
fmt.Printf("EVENT: %X\n", event.Event) fmt.Printf("EVENT: %s\n", event.Event)
fmt.Printf("DATA: %X\n", event.Data) fmt.Printf("DATA: %s\n", event.Data)
*/
if event.Data != nil { if event.Data != nil {
notifyResponse := &NotifyResponse{} notifyResponse := &NotifyResponse{}
@ -72,28 +70,19 @@ func (e *EventStream) Listen() {
break break
} }
fmt.Printf("%s\n", notifyResponse)
if notifyResponse.Status == "connected" { if notifyResponse.Status == "connected" {
fmt.Println("Connected.")
e.Connected = true e.Connected = true
} else if notifyResponse.Status == "disconnected" { } else if notifyResponse.Status == "disconnected" {
fmt.Println("Disconnected.")
e.Connected = false e.Connected = false
} else { } else {
fmt.Printf("Message for transId: %s\n", notifyResponse.TransId)
if subscriber, ok := e.Subscriptions[notifyResponse.TransId]; ok { if subscriber, ok := e.Subscriptions[notifyResponse.TransId]; ok {
e.Lock() e.Lock()
*subscriber <- *notifyResponse subscriber <- notifyResponse
close(*subscriber) close(subscriber)
delete(e.Subscriptions, notifyResponse.TransId) delete(e.Subscriptions, notifyResponse.TransId)
e.Unlock() e.Unlock()
} else {
// Throw away the message.
fmt.Println("Throwing away message.")
} }
} }
} else {
fmt.Printf("Event data was nil.\n")
} }
} }
}() }()

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
import ( import (
"time" "time"

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
// UpdateResponse is an intermediate struct used when parsing data from the UpdateProfile() call. // UpdateResponse is an intermediate struct used when parsing data from the UpdateProfile() call.
type Status struct { type Status struct {

View File

@ -1,4 +1,4 @@
package arlo_golang package arlo
/* /*
// Credentials is the login credential data. // Credentials is the login credential data.