Tibber Pulse: switch to graphql-ws websocket transport (#5808)
This commit is contained in:
parent
e229f7ed2e
commit
938d47ca4f
3 changed files with 54 additions and 38 deletions
2
go.mod
2
go.mod
|
|
@ -47,7 +47,7 @@ require (
|
|||
github.com/gregdel/pushover v1.1.0
|
||||
github.com/grid-x/modbus v0.0.0-20230105075906-8b31e0d1647e
|
||||
github.com/hashicorp/go-version v1.6.0
|
||||
github.com/hasura/go-graphql-client v0.8.1
|
||||
github.com/hasura/go-graphql-client v0.8.2-0.20230118111322-12655772463c
|
||||
github.com/imdario/mergo v0.3.13
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.12.1
|
||||
github.com/itchyny/gojq v0.12.11
|
||||
|
|
|
|||
4
go.sum
4
go.sum
|
|
@ -441,8 +441,8 @@ github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO
|
|||
github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
|
||||
github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
|
||||
github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
|
||||
github.com/hasura/go-graphql-client v0.8.1 h1:yU4888urgkW4L47cs+QQDXl3YfVaNraUqym5qsJ41Ms=
|
||||
github.com/hasura/go-graphql-client v0.8.1/go.mod h1:NVifIwv+YFIUYGLQ7SM2/vBbzS/9rFP4vmIf/vf/zXM=
|
||||
github.com/hasura/go-graphql-client v0.8.2-0.20230118111322-12655772463c h1:BruXcowJs5r1MKf534ObCBjA9zk5XKUZ5W2m4NdE3Ak=
|
||||
github.com/hasura/go-graphql-client v0.8.2-0.20230118111322-12655772463c/go.mod h1:NVifIwv+YFIUYGLQ7SM2/vBbzS/9rFP4vmIf/vf/zXM=
|
||||
github.com/hinshun/vt10x v0.0.0-20220119200601-820417d04eec h1:qv2VnGeEQHchGaZ/u7lxST/RaJw+cv273q79D81Xbog=
|
||||
github.com/hinshun/vt10x v0.0.0-20220119200601-820417d04eec/go.mod h1:Q48J4R4DvxnHolD5P8pOtXigYlRuPLGl6moFx3ulM68=
|
||||
github.com/holoplot/go-avahi v1.0.1 h1:XcqR2keL4qWRnlxHD5CAOdWpLFZJ+EOUK0vEuylfvvk=
|
||||
|
|
|
|||
|
|
@ -1,14 +1,18 @@
|
|||
package meter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/evcc-io/evcc/api"
|
||||
"github.com/evcc-io/evcc/meter/tibber"
|
||||
"github.com/evcc-io/evcc/util"
|
||||
"github.com/evcc-io/evcc/util/request"
|
||||
"github.com/evcc-io/evcc/util/transport"
|
||||
"github.com/hasura/go-graphql-client"
|
||||
)
|
||||
|
||||
|
|
@ -43,19 +47,43 @@ func NewTibberFromConfig(other map[string]interface{}) (api.Meter, error) {
|
|||
log: util.NewLogger("pulse").Redact(cc.Token, cc.HomeID),
|
||||
}
|
||||
|
||||
if cc.HomeID == "" {
|
||||
// query client
|
||||
qclient := tibber.NewClient(t.log, cc.Token)
|
||||
// query client
|
||||
qclient := tibber.NewClient(t.log, cc.Token)
|
||||
|
||||
home, err := qclient.DefaultHome("")
|
||||
if err != nil {
|
||||
if cc.HomeID == "" {
|
||||
if home, err := qclient.DefaultHome(""); err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
cc.HomeID = home.ID
|
||||
}
|
||||
cc.HomeID = home.ID
|
||||
}
|
||||
|
||||
var res struct {
|
||||
Viewer struct {
|
||||
WebsocketSubscriptionUrl string
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), request.Timeout)
|
||||
defer cancel()
|
||||
|
||||
if err := qclient.Query(ctx, &res, nil); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// subscription client
|
||||
client := graphql.NewSubscriptionClient(tibber.SubscriptionURI).
|
||||
client := graphql.NewSubscriptionClient(res.Viewer.WebsocketSubscriptionUrl).
|
||||
WithProtocol(graphql.GraphQLWS).
|
||||
WithWebSocketOptions(graphql.WebsocketOptions{
|
||||
HTTPClient: &http.Client{
|
||||
Transport: &transport.Decorator{
|
||||
Base: http.DefaultTransport,
|
||||
Decorator: transport.DecorateHeaders(map[string]string{
|
||||
"User-Agent": "go-graphql-client/0.9.0",
|
||||
}),
|
||||
},
|
||||
},
|
||||
}).
|
||||
WithConnectionParams(map[string]any{
|
||||
"token": cc.Token,
|
||||
}).
|
||||
|
|
@ -63,36 +91,26 @@ func NewTibberFromConfig(other map[string]interface{}) (api.Meter, error) {
|
|||
WithLog(t.log.TRACE.Println)
|
||||
|
||||
// run the client
|
||||
go func() {
|
||||
if err := client.Run(); err != nil {
|
||||
t.log.ERROR.Println(err)
|
||||
}
|
||||
}()
|
||||
|
||||
err := t.subscribe(client, cc.HomeID)
|
||||
done := make(chan error)
|
||||
go t.subscribe(client, cc.HomeID, done)
|
||||
err := <-done
|
||||
|
||||
return t, err
|
||||
}
|
||||
|
||||
// subscribe to the websocket query
|
||||
func (t *Tibber) subscribe(client *graphql.SubscriptionClient, homeID string) error {
|
||||
func (t *Tibber) subscribe(client *graphql.SubscriptionClient, homeID string, done chan error) {
|
||||
var query struct {
|
||||
tibber.LiveMeasurement `graphql:"liveMeasurement(homeId: $homeId)"`
|
||||
}
|
||||
|
||||
var once sync.Once
|
||||
recv := make(chan struct{})
|
||||
errC := make(chan error)
|
||||
|
||||
_, err := client.Subscribe(&query, map[string]any{
|
||||
"homeId": graphql.ID(homeID),
|
||||
}, func(data []byte, err error) error {
|
||||
if err != nil {
|
||||
select {
|
||||
case errC <- err:
|
||||
default:
|
||||
}
|
||||
return nil
|
||||
once.Do(func() { done <- err })
|
||||
}
|
||||
|
||||
var res struct {
|
||||
|
|
@ -100,33 +118,31 @@ func (t *Tibber) subscribe(client *graphql.SubscriptionClient, homeID string) er
|
|||
}
|
||||
|
||||
if err := json.Unmarshal(data, &res); err != nil {
|
||||
once.Do(func() { done <- err })
|
||||
|
||||
t.log.ERROR.Println(err)
|
||||
return nil
|
||||
}
|
||||
|
||||
once.Do(func() {
|
||||
close(recv)
|
||||
})
|
||||
|
||||
t.mu.Lock()
|
||||
t.live = res.LiveMeasurement
|
||||
t.updated = time.Now()
|
||||
t.mu.Unlock()
|
||||
|
||||
once.Do(func() { close(done) })
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// wait for connection
|
||||
if err == nil {
|
||||
select {
|
||||
case <-recv:
|
||||
case <-time.After(timeout):
|
||||
err = api.ErrTimeout
|
||||
case err = <-errC:
|
||||
}
|
||||
if err != nil {
|
||||
once.Do(func() { done <- err })
|
||||
}
|
||||
|
||||
return err
|
||||
go func() {
|
||||
if err := client.Run(); err != nil {
|
||||
once.Do(func() { done <- err })
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// CurrentPower implements the api.Meter interface
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue