updates

config/config.go

@@ -1055,14 +1055,20 @@ type TxIndexConfig struct {
 	//
 	// Options:
 	//   1) "null"
-	//   2) "kv" (default) - the simplest possible indexer,
+	//   2) "kv" (default) - The simplest possible indexer,
 	//      backed by key-value storage (defaults to levelDB; see DBBackend).
-	//   3) "psql" - the indexer services backed by PostgreSQL.
+	//   3) "psql" - The indexer services backed by PostgreSQL.
+	//   4) "pubsub" - The indexer services backed by Google Cloud PubSub.
 	Indexer string `mapstructure:"indexer"`
 
 	// The PostgreSQL connection configuration, the connection format:
 	// postgresql://<user>:<password>@<host>:<port>/<db>?<opts>
 	PsqlConn string `mapstructure:"psql-conn"`
+
+	// PubsubProjectID defines the Google Cloud Pubsub project ID. Note, operators
+	// must ensure the GOOGLE_APPLICATION_CREDENTIALS environment variable is set
+	// to the location of their creds file.
+	PubsubProjectID string `mapstructure:"pubsub-project-id"`
 }
 
 // DefaultTxIndexConfig returns a default configuration for the transaction indexer.

config/toml.go

@@ -471,9 +471,10 @@ peer_query_maj23_sleep_duration = "{{ .Consensus.PeerQueryMaj23SleepDuration }}"
 #
 # Options:
 #   1) "null"
-#   2) "kv" (default) - the simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend).
+#   2) "kv" (default) - The simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend).
 # 		- When "kv" is chosen "tx.height" and "tx.hash" will always be indexed.
-#   3) "psql" - the indexer services backed by PostgreSQL.
+#   3) "psql" - The indexer services backed by PostgreSQL.
+#   4) "pubsub" - The indexer services backed by Google Cloud Pubsub.
 # When "kv" or "psql" is chosen "tx.height" and "tx.hash" will always be indexed.
 indexer = "{{ .TxIndex.Indexer }}"
 
@@ -481,6 +482,11 @@ indexer = "{{ .TxIndex.Indexer }}"
 #   postgresql://<user>:<password>@<host>:<port>/<db>?<opts>
 psql-conn = "{{ .TxIndex.PsqlConn }}"
 
+# The Google Cloud Pubsub project ID. Note, operators must ensure the
+# GOOGLE_APPLICATION_CREDENTIALS environment variable is set to the location of
+# their creds file.
+pubsub-project-id = "{{ .TxIndex.PubsubProjectID }}"
+
 #######################################################
 ###       Instrumentation Configuration Options     ###
 #######################################################

state/indexer/sink/kafka/kafka.go

@@ -1,17 +0,0 @@
-package kafka
-
-import (
-	"cloud.google.com/go/pubsub"
-
-	"github.com/tendermint/tendermint/state/indexer"
-	"github.com/tendermint/tendermint/state/txindex"
-)
-
-var (
-	_ indexer.BlockIndexer = (*EventSink)(nil)
-	_ txindex.TxIndexer    = (*EventSink)(nil)
-)
-
-type EventSink struct {
-	client *pubsub.Client
-}

state/indexer/sink/pubsub/pubsub.go

@@ -0,0 +1,44 @@
+package pubsub
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"cloud.google.com/go/pubsub"
+
+	"github.com/tendermint/tendermint/state/indexer"
+	"github.com/tendermint/tendermint/state/txindex"
+)
+
+const credsEnvVar = "GOOGLE_APPLICATION_CREDENTIALS"
+
+var (
+	_ indexer.BlockIndexer = (*EventSink)(nil)
+	_ txindex.TxIndexer    = (*EventSink)(nil)
+)
+
+type EventSink struct {
+	client  *pubsub.Client
+	chainID string
+}
+
+func NewEventSink(connStr, chainID string) (*EventSink, error) {
+	if s := os.Getenv(credsEnvVar); len(s) == 0 {
+		return nil, fmt.Errorf("missing '%s' environment variable", credsEnvVar)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	c, err := pubsub.NewClient(ctx, "project-id")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Google Cloud Pubsub client: %w", err)
+	}
+
+	return &EventSink{
+		client:  c,
+		chainID: chainID,
+	}, nil
+}