func (svc *Kinesis) Errorf(format string, v ...interface{}) {
	svc.logger.Errorf(serviceName, format, v...)
}
Errorf logging error information.
package kinesis
import (
	"strings"
	"sync"
	"github.com/aws/aws-sdk-go/aws/session"
	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/pointers"
)
const (
	serviceName = "Kinesis"
)
type Kinesis struct {
	client *SDK.Kinesis
	logger log.Logger
	prefix string
	streamsMu sync.RWMutex
	streams   map[string]*Stream
}
func New(conf config.Config) (*Kinesis, error) {
	sess, err := conf.Session()
	if err != nil {
		return nil, err
	}
	svc := NewFromSession(sess)
	svc.prefix = conf.DefaultPrefix
	return svc, nil
}
func NewFromSession(sess *session.Session) *Kinesis {
	return &Kinesis{
		client:  SDK.New(sess),
		logger:  log.DefaultLogger,
		streams: make(map[string]*Stream),
	}
}
func (svc *Kinesis) GetClient() *SDK.Kinesis {
	return svc.client
}
func (svc *Kinesis) SetLogger(logger log.Logger) {
	svc.logger = logger
}
func (svc *Kinesis) SetPrefix(prefix string) {
	svc.prefix = prefix
}
func (svc *Kinesis) GetStream(name string) (*Stream, error) {
	streamName := svc.prefix + name
	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
}
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
}
func (svc *Kinesis) CreateStreamWithName(name string) error {
	streamName := svc.prefix + name
	return svc.CreateStream(&SDK.CreateStreamInput{
		StreamName: pointers.String(streamName),
		ShardCount: pointers.Long(1),
	})
}
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:  
		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)
}
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
}
func (svc *Kinesis) Infof(format string, v ...interface{}) {
	svc.logger.Infof(serviceName, format, v...)
}