client.go

package kinesis

import (
	"strings"
	"sync"

	SDK "github.com/aws/aws-sdk-go/service/kinesis"

	"github.com/evalphobia/aws-sdk-go-wrapper/config"
	"github.com/evalphobia/aws-sdk-go-wrapper/log"
	"github.com/evalphobia/aws-sdk-go-wrapper/private/errors"
	"github.com/evalphobia/aws-sdk-go-wrapper/private/pointers"
)

const (
	serviceName = "Kinesis"
)

// Kinesis has Kinesis client.
type Kinesis struct {
	client *SDK.Kinesis

	logger log.Logger
	prefix string

	streamsMu sync.RWMutex
	streams   map[string]*Stream
}

// New returns initialized *Kinesis.
func New(conf config.Config) (*Kinesis, error) {
	sess, err := conf.Session()
	if err != nil {
		return nil, err
	}

	svc := &Kinesis{
		client:  SDK.New(sess),
		logger:  log.DefaultLogger,
		prefix:  conf.DefaultPrefix,
		streams: make(map[string]*Stream),
	}
	return svc, nil
}

// SetLogger sets logger.
func (svc *Kinesis) SetLogger(logger log.Logger) {
	svc.logger = logger
}

// GetStream gets Kinesis Stream.
func (svc *Kinesis) GetStream(name string) (*Stream, error) {
	streamName := svc.prefix + name

	// get the stream from cache
	svc.streamsMu.RLock()
	s, ok := svc.streams[streamName]
	svc.streamsMu.RUnlock()
	if ok {
		return s, nil
	}

	s, err := NewStream(svc, name)
	if err != nil {
		return nil, err
	}

	svc.streamsMu.Lock()
	svc.streams[streamName] = s
	svc.streamsMu.Unlock()
	return s, nil
}

// CreateStream creates new Kinesis Stream.
func (svc *Kinesis) CreateStream(in *SDK.CreateStreamInput) error {
	_, err := svc.client.CreateStream(in)
	if err != nil {
		svc.Errorf("error on `CreateStream` operation; stream=%s; error=%s;", *in.StreamName, err.Error())
		return err
	}

	svc.Infof("success on `CreateStream` operation; stream=%s;", *in.StreamName)
	return nil
}

// CreateStreamWithName creates new Kinesis Stream by given name with prefix.
func (svc *Kinesis) CreateStreamWithName(name string) error {
	streamName := svc.prefix + name
	return svc.CreateStream(&SDK.CreateStreamInput{
		StreamName: pointers.String(streamName),
		ShardCount: pointers.Long(1),
	})
}

// IsExistStream checks if the Stream already exists or not.
func (svc *Kinesis) IsExistStream(name string) (bool, error) {
	streamName := svc.prefix + name
	desc, err := svc.client.DescribeStream(&SDK.DescribeStreamInput{
		StreamName: pointers.String(streamName),
	})

	switch {
	case isNonExistentStreamError(err):
		return false, nil
	case err != nil:
		svc.Errorf("error on `DescribeStream` operation; stream=%s; error=%s", streamName, err.Error())
		return false, err
	case desc == nil:
		return false, nil
	case desc.StreamDescription != nil: // stream exists
		return true, nil
	default:
		return false, nil
	}
}

func isNonExistentStreamError(err error) bool {
	const errNonExistentStream = "ResourceNotFoundException: "
	if err == nil {
		return false
	}

	return strings.Contains(err.Error(), errNonExistentStream)
}

// ForceDeleteStream deletes Kinesis stream by given name with prefix.
func (svc *Kinesis) ForceDeleteStream(name string) error {
	streamName := svc.prefix + name
	_, err := svc.client.DeleteStream(&SDK.DeleteStreamInput{
		StreamName: pointers.String(streamName),
	})
	if err != nil {
		svc.Errorf("error on `DeleteStream` operation; stream=%s; error=%s;", streamName, err.Error())
		return err
	}

	svc.Infof("success on `DeleteStream` operation; stream=%s;", streamName)
	return nil
}

// Infof logging information.
func (svc *Kinesis) Infof(format string, v ...interface{}) {
	svc.logger.Infof(serviceName, format, v...)
}

// Errorf logging error information.
func (svc *Kinesis) Errorf(format string, v ...interface{}) {
	svc.logger.Errorf(serviceName, format, v...)
}

func newErrors() *errors.Errors {
	return errors.NewErrors(serviceName)
}