func NewAerospikeSource(ctx context.Context,
	properties *AerospikeProperties,
	scanPolicy *aero.ScanPolicy,
	changeNotificationProperties *ChangeNotificationProperties) (*AerospikeSource, error) {
	client, err := aero.NewClientWithPolicy(properties.Policy, properties.Hostname, properties.Port)
	if err != nil {
		return nil, err
	}
	if scanPolicy == nil {
		scanPolicy = aero.NewScanPolicy()
	}
	records := make(chan *aero.Result)
	source := &AerospikeSource{
		client:                       client,
		records:                      records,
		scanPolicy:                   scanPolicy,
		out:                          make(chan interface{}),
		ctx:                          ctx,
		properties:                   properties,
		changeNotificationProperties: changeNotificationProperties,
	}
	go source.poll()
	go source.init()
	return source, nil
}
NewAerospikeSource returns a new AerospikeSource instance
set changeNotificationProperties to nil to scan the entire namespace/set
package aerospike
import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"
	aero "github.com/aerospike/aerospike-client-go"
	"github.com/reugn/go-streams"
	"github.com/reugn/go-streams/flow"
)
type AerospikeProperties struct {
	Policy    *aero.ClientPolicy
	Hostname  string
	Port      int
	Namespase string
	SetName   string
}
type ChangeNotificationProperties struct {
	PollingInterval time.Duration
}
type AerospikeSource struct {
	client                       *aero.Client
	records                      chan *aero.Result
	scanPolicy                   *aero.ScanPolicy
	out                          chan interface{}
	ctx                          context.Context
	properties                   *AerospikeProperties
	changeNotificationProperties *ChangeNotificationProperties
}