Export shim publisher functions

Kathryn Baldauf · Kathryn Baldauf · commit 2d8a65b1b244 · 2019-08-27T17:15:15.000-07:00
- Our out of tree shim would like to publish events with ttrpc. These
functions should be exposed so our shim doesn't need to reimplement
publisher logic.

Signed-off-by: Kathryn Baldauf &lt;kabaldau@microsoft.com&gt;
diff --git a/runtime/v2/shim/publisher.go b/runtime/v2/shim/publisher.go
@@ -41,13 +41,13 @@ type item struct {
 	count int
 }
 
-func newPublisher(address string) (*remoteEventsPublisher, error) {
+func NewPublisher(address string) (*RemoteEventsPublisher, error) {
 	client, err := ttrpcutil.NewClient(address)
 	if err != nil {
 		return nil, err
 	}
 
-	l := &remoteEventsPublisher{
+	l := &RemoteEventsPublisher{
 		client:  client,
 		closed:  make(chan struct{}),
 		requeue: make(chan *item, queueSize),
@@ -57,26 +57,26 @@ func newPublisher(address string) (*remoteEventsPublisher, error) {
 	return l, nil
 }
 
-type remoteEventsPublisher struct {
+type RemoteEventsPublisher struct {
 	client  *ttrpcutil.Client
 	closed  chan struct{}
 	closer  sync.Once
 	requeue chan *item
 }
 
-func (l *remoteEventsPublisher) Done() <-chan struct{} {
+func (l *RemoteEventsPublisher) Done() <-chan struct{} {
 	return l.closed
 }
 
-func (l *remoteEventsPublisher) Close() (err error) {
+func (l *RemoteEventsPublisher) Close() (err error) {
 	err = l.client.Close()
 	l.closer.Do(func() {
 		close(l.closed)
 	})
 	return err
 }
 
-func (l *remoteEventsPublisher) processQueue() {
+func (l *RemoteEventsPublisher) processQueue() {
 	for i := range l.requeue {
 		if i.count > maxRequeue {
 			logrus.Errorf("evicting %s from queue because of retry count", i.ev.Topic)
@@ -91,7 +91,7 @@ func (l *remoteEventsPublisher) processQueue() {
 	}
 }
 
-func (l *remoteEventsPublisher) queue(i *item) {
+func (l *RemoteEventsPublisher) queue(i *item) {
 	go func() {
 		i.count++
 		// re-queue after a short delay
@@ -100,7 +100,7 @@ func (l *remoteEventsPublisher) queue(i *item) {
 	}()
 }
 
-func (l *remoteEventsPublisher) Publish(ctx context.Context, topic string, event events.Event) error {
+func (l *RemoteEventsPublisher) Publish(ctx context.Context, topic string, event events.Event) error {
 	ns, err := namespaces.NamespaceRequired(ctx)
 	if err != nil {
 		return err
@@ -127,7 +127,7 @@ func (l *remoteEventsPublisher) Publish(ctx context.Context, topic string, event
 	return nil
 }
 
-func (l *remoteEventsPublisher) forwardRequest(ctx context.Context, req *v1.ForwardRequest) error {
+func (l *RemoteEventsPublisher) forwardRequest(ctx context.Context, req *v1.ForwardRequest) error {
 	_, err := l.client.EventsService().Forward(ctx, req)
 	if err == nil {
 		return nil
diff --git a/runtime/v2/shim/shim.go b/runtime/v2/shim/shim.go
@@ -169,7 +169,7 @@ func run(id string, initFunc Init, config Config) error {
 
 	ttrpcAddress := os.Getenv(ttrpcAddressEnv)
 
-	publisher, err := newPublisher(ttrpcAddress)
+	publisher, err := NewPublisher(ttrpcAddress)
 	if err != nil {
 		return err
 	}

Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,7 @@ func run(id string, initFunc Init, config Config) error {`
`169`	`169`
`170`	`170`	`ttrpcAddress := os.Getenv(ttrpcAddressEnv)`
`171`	`171`
`172`		`- publisher, err := newPublisher(ttrpcAddress)`
	`172`	`+ publisher, err := NewPublisher(ttrpcAddress)`
`173`	`173`	`if err != nil {`
`174`	`174`	`return err`
`175`	`175`	`}`