package provider import ( "context" "math" "net/http" "strconv" "sync" "time" "github.com/evcc-io/evcc/api" "github.com/evcc-io/evcc/provider/pipeline" "github.com/evcc-io/evcc/util" "github.com/evcc-io/evcc/util/request" "github.com/evcc-io/evcc/util/transport" "nhooyr.io/websocket" ) const retryDelay = 5 * time.Second // Socket implements websocket request provider type Socket struct { *request.Helper log *util.Logger url string headers map[string]string scale float64 pipeline *pipeline.Pipeline val *util.Monitor[[]byte] } func init() { registry.Add("ws", NewSocketProviderFromConfig) registry.Add("websocket", NewSocketProviderFromConfig) } // NewSocketProviderFromConfig creates a HTTP provider func NewSocketProviderFromConfig(other map[string]interface{}) (Provider, error) { cc := struct { URI string Headers map[string]string pipeline.Settings `mapstructure:",squash"` Scale float64 Insecure bool Auth Auth Timeout time.Duration }{ Headers: make(map[string]string), Scale: 1, } if err := util.DecodeOther(other, &cc); err != nil { return nil, err } log := util.NewLogger("ws") url := util.DefaultScheme(cc.URI, "ws") if url != cc.URI { log.WARN.Printf("missing scheme for %s, assuming ws", cc.URI) } p := &Socket{ log: log, Helper: request.NewHelper(log), url: url, headers: cc.Headers, scale: cc.Scale, val: util.NewMonitor[[]byte](cc.Timeout), } // handle basic auth if cc.Auth.Type != "" { basicAuth := transport.BasicAuthHeader(cc.Auth.User, cc.Auth.Password) log.Redact(basicAuth) p.headers["Authorization"] = basicAuth } // ignore the self signed certificate if cc.Insecure { p.Client.Transport = request.NewTripper(log, transport.Insecure()) } var err error if p.pipeline, err = pipeline.New(log, cc.Settings); err != nil { return nil, err } errC := make(chan error, 1) go p.run(errC) if cc.Timeout > 0 { select { case <-p.val.Done(): case <-time.After(cc.Timeout): return nil, api.ErrTimeout case err := <-errC: return nil, err } } return p, nil } func (p *Socket) run(errC chan error) { var once sync.Once headers := make(http.Header) for k, v := range p.headers { headers.Set(k, v) } opts := &websocket.DialOptions{ HTTPHeader: headers, } for { ctx, cancel := context.WithTimeout(context.Background(), request.Timeout) conn, _, err := websocket.Dial(ctx, p.url, opts) cancel() if err != nil { // handle initial connection error immediately once.Do(func() { errC <- err }) p.log.ERROR.Println(err) time.Sleep(retryDelay) continue } for { _, b, err := conn.Read(context.Background()) if err != nil { p.log.TRACE.Println("read:", err) _ = conn.Close(websocket.StatusAbnormalClosure, "done") break } p.log.TRACE.Printf("recv: %s", b) if v, err := p.pipeline.Process(b); err == nil { p.val.Set(v) } } } } var _ StringProvider = (*Socket)(nil) // StringGetter sends string request func (p *Socket) StringGetter() (func() (string, error), error) { return func() (string, error) { val, err := p.val.Get() return string(val), err }, nil } var _ FloatProvider = (*Socket)(nil) // FloatGetter parses float from string getter func (p *Socket) FloatGetter() (func() (float64, error), error) { g, err := p.StringGetter() return func() (float64, error) { s, err := g() if err != nil { return 0, err } f, err := strconv.ParseFloat(s, 64) return f * p.scale, err }, err } var _ IntProvider = (*Socket)(nil) // IntGetter parses int64 from float getter func (p *Socket) IntGetter() (func() (int64, error), error) { g, err := p.FloatGetter() return func() (int64, error) { f, err := g() return int64(math.Round(f)), err }, err } var _ BoolProvider = (*Socket)(nil) // BoolGetter parses bool from string getter func (p *Socket) BoolGetter() (func() (bool, error), error) { g, err := p.StringGetter() return func() (bool, error) { s, err := g() return util.Truish(s), err }, err }