Moving to cev2 - publisher, receive adapter - Implementing KRShaped

cmd/broker/ingress/main.go

@@ -17,11 +17,11 @@ limitations under the License.
 package main
 
 import (
-	"github.com/google/knative-gcp/pkg/broker/ingress"
 	metadataClient "github.com/google/knative-gcp/pkg/gclient/metadata"
 	"github.com/google/knative-gcp/pkg/metrics"
 	"github.com/google/knative-gcp/pkg/utils"
 	"github.com/google/knative-gcp/pkg/utils/appcredentials"
+	"github.com/google/knative-gcp/pkg/utils/clients"
 	"github.com/google/knative-gcp/pkg/utils/mainhelper"
 
 	"go.uber.org/zap"
@@ -59,8 +59,8 @@ func main() {
 
 	ingress, err := InitializeHandler(
 		ctx,
-		ingress.Port(env.Port),
-		ingress.ProjectID(projectID),
+		clients.Port(env.Port),
+		clients.ProjectID(projectID),
 		metrics.PodName(env.PodName),
 		metrics.ContainerName(component),
 	)

cmd/broker/ingress/wire.go

@@ -24,13 +24,14 @@ import (
 	"github.com/google/knative-gcp/pkg/broker/config/volume"
 	"github.com/google/knative-gcp/pkg/broker/ingress"
 	"github.com/google/knative-gcp/pkg/metrics"
+	"github.com/google/knative-gcp/pkg/utils/clients"
 	"github.com/google/wire"
 )
 
 func InitializeHandler(
 	ctx context.Context,
-	port ingress.Port,
-	projectID ingress.ProjectID,
+	port clients.Port,
+	projectID clients.ProjectID,
 	podName metrics.PodName,
 	containerName metrics.ContainerName,
 ) (*ingress.Handler, error) {

cmd/pubsub/publisher/main.go

@@ -21,17 +21,22 @@ import (
 	"flag"
 	"log"
 
-	"cloud.google.com/go/compute/metadata"
+	metadataClient "github.com/google/knative-gcp/pkg/gclient/metadata"
+	. "github.com/google/knative-gcp/pkg/pubsub/publisher"
+	tracingconfig "github.com/google/knative-gcp/pkg/tracing"
+	"github.com/google/knative-gcp/pkg/utils"
+	"github.com/google/knative-gcp/pkg/utils/appcredentials"
+	"github.com/google/knative-gcp/pkg/utils/clients"
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"knative.dev/eventing/pkg/tracing"
-
-	"github.com/google/knative-gcp/pkg/pubsub/publisher"
-	tracingconfig "github.com/google/knative-gcp/pkg/tracing"
 )
 
 type envConfig struct {
+	// Environment variable containing the port for the publisher.
+	Port int `envconfig:"PORT" default:"8080"`
+
 	// Environment variable containing project id.
 	Project string `envconfig:"PROJECT_ID"`
 
@@ -47,6 +52,8 @@ type envConfig struct {
 }
 
 func main() {
+	appcredentials.MustExistOrUnsetEnv()
+
 	flag.Parse()
 
 	ctx := context.Background()
@@ -62,15 +69,15 @@ func main() {
 		logger.Fatal("Failed to process env var", zap.Error(err))
 	}
 
-	if env.Project == "" {
-		project, err := metadata.ProjectID()
-		if err != nil {
-			logger.Fatal("failed to find project id. ", zap.Error(err))
-		}
-		env.Project = project
+	projectID, err := utils.ProjectID(env.Project, metadataClient.NewDefaultMetadataClient())
+	if err != nil {
+		logger.Fatal("Failed to retrieve project id", zap.Error(err))
 	}
 
-	logger.Info("Using project.", zap.String("project", env.Project))
+	topicID := env.Topic
+	if topicID == "" {
+		logger.Fatal("Failed to retrieve topic id", zap.Error(err))
+	}
 
 	tracingConfig, err := tracingconfig.JSONToConfig(env.TracingConfigJson)
 	if err != nil {
@@ -80,13 +87,21 @@ func main() {
 		logger.Error("Failed to setup tracing", zap.Error(err), zap.Any("tracingConfig", tracingConfig))
 	}
 
-	startable := &publisher.Publisher{
-		ProjectID: env.Project,
-		TopicID:   env.Topic,
+	logger.Info("Initializing publisher", zap.String("Project ID", projectID), zap.String("Topic ID", topicID))
+
+	publisher, err := InitializePublisher(
+		ctx,
+		clients.Port(env.Port),
+		clients.ProjectID(projectID),
+		TopicID(topicID),
+	)
+
+	if err != nil {
+		logger.Fatal("Unable to create publisher", zap.Error(err))
 	}
 
-	logger.Info("Starting Pub/Sub Publisher.", zap.Any("publisher", startable))
-	if err := startable.Start(ctx); err != nil {
-		logger.Fatal("failed to start publisher: ", zap.Error(err))
+	logger.Info("Starting publisher", zap.Any("publisher", publisher))
+	if err := publisher.Start(ctx); err != nil {
+		logger.Fatal("Failed to start publisher", zap.Error(err))
 	}
 }

cmd/pubsub/publisher/wire.go

@@ -0,0 +1,39 @@
+// +build wireinject
+
+/*
+Copyright 2020 Google LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+
+	"github.com/google/knative-gcp/pkg/pubsub/publisher"
+	"github.com/google/knative-gcp/pkg/utils/clients"
+
+	"github.com/google/wire"
+)
+
+func InitializePublisher(
+	ctx context.Context,
+	port clients.Port,
+	projectID clients.ProjectID,
+	topicID publisher.TopicID,
+) (*publisher.Publisher, error) {
+	panic(wire.Build(
+		publisher.PublisherSet,
+	))
+}