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
}
CreateStream creates new Kinesis Stream.
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
}