func_context.go 1.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. // Copyright 2021 EMQ Technologies Co., Ltd.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package context
  15. import (
  16. "fmt"
  17. "github.com/lf-edge/ekuiper/internal/topo/connection/clients"
  18. "github.com/lf-edge/ekuiper/pkg/api"
  19. )
  20. type DefaultFuncContext struct {
  21. api.StreamContext
  22. funcId int
  23. }
  24. func NewDefaultFuncContext(ctx api.StreamContext, id int) *DefaultFuncContext {
  25. return &DefaultFuncContext{
  26. StreamContext: ctx,
  27. funcId: id,
  28. }
  29. }
  30. func (c *DefaultFuncContext) IncrCounter(key string, amount int) error {
  31. return c.StreamContext.IncrCounter(c.convertKey(key), amount)
  32. }
  33. func (c *DefaultFuncContext) GetCounter(key string) (int, error) {
  34. return c.StreamContext.GetCounter(c.convertKey(key))
  35. }
  36. func (c *DefaultFuncContext) PutState(key string, value interface{}) error {
  37. return c.StreamContext.PutState(c.convertKey(key), value)
  38. }
  39. func (c *DefaultFuncContext) GetState(key string) (interface{}, error) {
  40. return c.StreamContext.GetState(c.convertKey(key))
  41. }
  42. func (c *DefaultFuncContext) DeleteState(key string) error {
  43. return c.StreamContext.DeleteState(c.convertKey(key))
  44. }
  45. func (c *DefaultFuncContext) GetFuncId() int {
  46. return c.funcId
  47. }
  48. func (c *DefaultFuncContext) GetClient(clientType string, config map[string]interface{}) (api.MessageClient, error) {
  49. return clients.GetClient(clientType, config)
  50. }
  51. func (c *DefaultFuncContext) convertKey(key string) string {
  52. return fmt.Sprintf("$$func%d_%s", c.funcId, key)
  53. }