Mainflux.mainflux/things/redis/streams.go

//
// Copyright (c) 2018
// Mainflux
//
// SPDX-License-Identifier: Apache-2.0
//

package redis

import (
	"github.com/go-redis/redis"
	"github.com/mainflux/mainflux/things"
)

const (
	streamID  = "mainflux.things"
	streamLen = 1000
)

var _ things.Service = (*eventStore)(nil)

type eventStore struct {
	svc    things.Service
	client *redis.Client
}

// NewEventStoreMiddleware returns wrapper around things service that sends
// events to event store.
func NewEventStoreMiddleware(svc things.Service, client *redis.Client) things.Service {
	return eventStore{
		svc:    svc,
		client: client,
	}
}

func (es eventStore) AddThing(key string, thing things.Thing) (things.Thing, error) {
	sth, err := es.svc.AddThing(key, thing)
	if err != nil {
		return sth, err
	}

	event := createThingEvent{
		id:       sth.ID,
		owner:    sth.Owner,
		kind:     sth.Type,
		name:     sth.Name,
		metadata: sth.Metadata,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return sth, err
}

func (es eventStore) UpdateThing(key string, thing things.Thing) error {
	if err := es.svc.UpdateThing(key, thing); err != nil {
		return err
	}

	event := updateThingEvent{
		id:       thing.ID,
		kind:     thing.Type,
		name:     thing.Name,
		metadata: thing.Metadata,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) ViewThing(key, id string) (things.Thing, error) {
	return es.svc.ViewThing(key, id)
}

func (es eventStore) ListThings(key string, offset, limit uint64) (things.ThingsPage, error) {
	return es.svc.ListThings(key, offset, limit)
}

func (es eventStore) ListThingsByChannel(key, id string, offset, limit uint64) (things.ThingsPage, error) {
	return es.svc.ListThingsByChannel(key, id, offset, limit)
}

func (es eventStore) RemoveThing(key, id string) error {
	if err := es.svc.RemoveThing(key, id); err != nil {
		return err
	}

	event := removeThingEvent{
		id: id,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) CreateChannel(key string, channel things.Channel) (things.Channel, error) {
	sch, err := es.svc.CreateChannel(key, channel)
	if err != nil {
		return sch, err
	}

	event := createChannelEvent{
		id:       sch.ID,
		owner:    sch.Owner,
		name:     sch.Name,
		metadata: sch.Metadata,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return sch, err
}

func (es eventStore) UpdateChannel(key string, channel things.Channel) error {
	if err := es.svc.UpdateChannel(key, channel); err != nil {
		return err
	}

	event := updateChannelEvent{
		id:       channel.ID,
		name:     channel.Name,
		metadata: channel.Metadata,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) ViewChannel(key, id string) (things.Channel, error) {
	return es.svc.ViewChannel(key, id)
}

func (es eventStore) ListChannels(key string, offset, limit uint64) (things.ChannelsPage, error) {
	return es.svc.ListChannels(key, offset, limit)
}

func (es eventStore) ListChannelsByThing(key, id string, offset, limit uint64) (things.ChannelsPage, error) {
	return es.svc.ListChannelsByThing(key, id, offset, limit)
}

func (es eventStore) RemoveChannel(key, id string) error {
	if err := es.svc.RemoveChannel(key, id); err != nil {
		return err
	}

	event := removeChannelEvent{
		id: id,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) Connect(key, chanID, thingID string) error {
	if err := es.svc.Connect(key, chanID, thingID); err != nil {
		return err
	}

	event := connectThingEvent{
		chanID:  chanID,
		thingID: thingID,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) Disconnect(key, chanID, thingID string) error {
	if err := es.svc.Disconnect(key, chanID, thingID); err != nil {
		return err
	}

	event := disconnectThingEvent{
		chanID:  chanID,
		thingID: thingID,
	}
	record := &redis.XAddArgs{
		Stream:       streamID,
		MaxLenApprox: streamLen,
		Values:       event.Encode(),
	}
	es.client.XAdd(record).Err()

	return nil
}

func (es eventStore) CanAccess(chanID string, key string) (string, error) {
	return es.svc.CanAccess(chanID, key)
}

func (es eventStore) Identify(key string) (string, error) {
	return es.svc.Identify(key)
}