shadow

Documentation

Index

• type BalanceStrategyCoPartitioned
  • func (strat *BalanceStrategyCoPartitioned) Name() string
  • func (strat *BalanceStrategyCoPartitioned) Plan(members map[string]sarama.ConsumerGroupMemberMetadata, ...) (sarama.BalanceStrategyPlan, error)
• type DeviceState
• type DeviceStateDB
• type DeviceStateMessage
• type Repo
  • func NewRedisRepo(addr string) (Repo, error)
• type Server
  • func (s *Server) Get(context context.Context, request *shadowpb.GetRequest) (response *shadowpb.GetResponse, err error)
  • func (s *Server) PatchDesiredState(context context.Context, request *shadowpb.PatchDesiredStateRequest) (response *shadowpb.PatchDesiredStateResponse, err error)
  • func (s *Server) StreamReportedStateChanges(request *shadowpb.StreamReportedStateChangesRequest, ...) (err error)
• type StateMerger
  • func (h *StateMerger) Cleanup(s sarama.ConsumerGroupSession) error
  • func (h *StateMerger) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *StateMerger) Setup(s sarama.ConsumerGroupSession) error

Types

type BalanceStrategyCoPartitioned

type BalanceStrategyCoPartitioned struct{}

func (*BalanceStrategyCoPartitioned) Name

func (strat *BalanceStrategyCoPartitioned) Name() string

func (*BalanceStrategyCoPartitioned) Plan

func (strat *BalanceStrategyCoPartitioned) Plan(members map[string]sarama.ConsumerGroupMemberMetadata, topics map[string][]int32) (sarama.BalanceStrategyPlan, error)

type DeviceState

type DeviceState struct {
	ID    string
	State DeviceStateMessage
}

TODO maybe use / embed DeviceStateMessage here - including Timestamp

type DeviceStateDB

type DeviceStateDB struct {
	ID              string
	ReportedVersion uint64
	ReportedState   postgres.Jsonb
	DesiredVersion  uint64
	DesiredState    postgres.Jsonb
}

type DeviceStateMessage

type DeviceStateMessage struct {
	Version   uint64
	State     json.RawMessage
	Timestamp time.Time
}

type Repo

type Repo interface {
	SetReported(DeviceState) (err error)
	GetReported(id string) (DeviceState, error)
	SetDesired(DeviceState) (err error)
	GetDesired(id string) (DeviceState, error)
}

func NewRedisRepo

func NewRedisRepo(addr string) (Repo, error)

type Server

type Server struct {
	Repo         Repo
	Producer     sarama.SyncProducer // Sync producer, we want to guarantee execution
	ProduceTopic string
	Log          *zap.Logger

	PubSub *pubsub.PubSub
}

Server is a data strcuture for shadow server

func (*Server) Get

func (s *Server) Get(context context.Context, request *shadowpb.GetRequest) (response *shadowpb.GetResponse, err error)

Get is a method to get a device state

func (*Server) PatchDesiredState

func (s *Server) PatchDesiredState(context context.Context, request *shadowpb.PatchDesiredStateRequest) (response *shadowpb.PatchDesiredStateResponse, err error)

PatchDesiredState is a method to patch a message to a device state

func (*Server) StreamReportedStateChanges

func (s *Server) StreamReportedStateChanges(request *shadowpb.StreamReportedStateChangesRequest, srv shadowpb.Shadows_StreamReportedStateChangesServer) (err error)

StreamReportedStateChanges is a method to start streaming of data from a device

type StateMerger

type StateMerger struct {
	SourceTopic    string // Incoming ticks
	MergedTopic    string // Full state with version
	RealDeltaTopic string // Deltas for each full state transition, with version

	ChangelogConsumerClient sarama.Client
	ChangelogProducerClient sarama.Client
	// contains filtered or unexported fields
}

func (*StateMerger) Cleanup

func (h *StateMerger) Cleanup(s sarama.ConsumerGroupSession) error

func (*StateMerger) ConsumeClaim

func (h *StateMerger) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*StateMerger) Setup

func (c *StateMerger) Setup(s sarama.ConsumerGroupSession) error