refactor: update test, rename variables & remove bin files

Signed-off-by: Imtiaz Uddin <imtiaz@appscode.com>

Author: Imtiaz246

apis/config.go

@@ -41,7 +41,7 @@ type ListenerCfg struct {
 type PublisherCfg struct {
 	Type            PublisherType `valid:"required" json:"type" mapstructure:"type"`
 	Address         string        `valid:"required" json:"address" mapstructure:"address"`
-	NatsCredPath    string        `json:"natsCredPath" mapstructure:"natsCredPath"`
+	NatsCredPath    string        `valid:"required" json:"natsCredPath" mapstructure:"natsCredPath"`
 	Topic           string        `valid:"required" json:"topic" mapstructure:"topic"`
 	TopicPrefix     string        `json:"topicPrefix" mapstructure:"topicPrefix"`
 	EnableTLS       bool          `json:"enableTLS" mapstructure:"enableTlS"`

apis/config_test.go

@@ -39,10 +39,11 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:        "kafka",
-					Address:     "addr",
-					Topic:       "stream",
-					TopicPrefix: "prefix",
+					Type:         "kafka",
+					Address:      "addr",
+					Topic:        "stream",
+					TopicPrefix:  "prefix",
+					NatsCredPath: "/etc/nats/creds/admin.creds",
 				},
 			},
 			wantErr: nil,
@@ -65,10 +66,11 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:        "kafka",
-					Address:     "addr",
-					Topic:       "stream",
-					TopicPrefix: "prefix",
+					Type:         "kafka",
+					Address:      "addr",
+					Topic:        "stream",
+					TopicPrefix:  "prefix",
+					NatsCredPath: "/etc/nats/creds/admin.creds",
 				},
 			},
 			wantErr: errors.New("Listener.refreshConnection: non zero value required;Listener.slotName: non zero value required"),
@@ -91,10 +93,11 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:        "kafka",
-					Address:     "addr",
-					Topic:       "stream",
-					TopicPrefix: "prefix",
+					Type:         "kafka",
+					Address:      "addr",
+					Topic:        "stream",
+					TopicPrefix:  "prefix",
+					NatsCredPath: "/etc/nats/creds/admin.creds",
 				},
 			},
 			wantErr: errors.New("Database.host: non zero value required;Database.port: non zero value required"),
@@ -119,8 +122,9 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Topic:       "stream",
-					TopicPrefix: "prefix",
+					Topic:        "stream",
+					TopicPrefix:  "prefix",
+					NatsCredPath: "/etc/nats/creds/admin.creds",
 				},
 			},
 			wantErr: errors.New("Publisher.address: non zero value required;Publisher.type: non zero value required"),

cmd/pgoutbox/main.go

@@ -85,12 +85,12 @@ func main() {
 
 			logger := apis.InitSlog(cfg.Logger, version, false)
 
-			conn, rConn, err := initPgxConnections(cfg.Database, logger, time.Minute*10)
+			pgxConn, pgConn, err := initPgxConnections(cfg.Database, logger, time.Minute*10)
 			if err != nil {
 				return fmt.Errorf("pgx connection: %w", err)
 			}
 
-			if err = configureReplicaIdentityToFull(ctx, conn, cfg.Listener.Filter); err != nil {
+			if err = configureReplicaIdentityToFull(ctx, pgxConn, cfg.Listener.Filter); err != nil {
 				return fmt.Errorf("configure replica identity: %w", err)
 			}
 			pub, err := factoryPublisher(ctx, cfg.Publisher, logger)
@@ -107,8 +107,8 @@ func main() {
 			svc := listener.NewWalListener(
 				cfg,
 				logger,
-				listener.NewRepository(conn),
-				newReplicationConn(rConn),
+				listener.NewRepository(pgxConn),
+				newReplicationConn(pgConn),
 				pub,
 				transaction.NewBinaryParser(logger, binary.BigEndian),
 				apis.NewMetrics(),

internal/listener/listener.go

@@ -143,6 +143,8 @@ func (l *Listener) liveness(w http.ResponseWriter, _ *http.Request) {
 		respCode = http.StatusInternalServerError
 
 		l.log.Warn("liveness probe failed")
+	} else {
+		l.log.Debug("liveness probe successful")
 	}
 
 	w.WriteHeader(respCode)
@@ -165,6 +167,8 @@ func (l *Listener) readiness(w http.ResponseWriter, _ *http.Request) {
 		respCode = http.StatusInternalServerError
 
 		l.log.Warn("readiness probe failed")
+	} else {
+		l.log.Debug("readiness probe successful")
 	}
 
 	w.WriteHeader(respCode)