Kafka consumer and Mortar

[ttyerl]

With our mortar service we also start a kafka consumer and dynamoDB.

func createApplication(configFilePath string, additionalFiles []string) *fx.App {
return fx.New(
		....
		mortar.ServiceAPIsAndOtherDependenciesFxOption(),
		mortar.DynamoFxOptions(),
		mortar.KafkaFxOption(),
		providers.BuildMortarWebServiceFxOption(),
	)

The kafkaConsumerImpl has a reference to Dynamo

kafkaConsumerImpl struct {
		deps              kafkaClientImplDeps
		kafkaReader       *kafka.Reader
		LazyDynamoClient  *dynamo.LazyDynamoClient
	}

The problem is within the kafka OnStart

OnStart: func(ctx context.Context) (startError error) {
			deps.Logger.Info(ctx, "CONSUMER --> Connecting to Kafka Cluster...")
			reader := kafka.NewReader(config)
			clientPtr.KafkaReader = &kafkaConsumerImpl{
				deps:        deps,
				kafkaReader: reader,
                                LazyDynamoClient: ........ // needs  reference from mortar.DynamoFxOptions()
			}
			return
		},

Can we use the ctx in to pass information between OnStart calls ?

[talgendler]

Making your DynamoClient lazy is a great choice in my opinion, but you should return a pointer to a wrapper.
For example:

type dynamoWrapper struct {
	LazyDynamoClient *dynamo.LazyDynamoClient
}

Make sure your Dynamo constructor returns a pointer to this struct

func CreateDynamoWrapper(lc fx.Lifecycle) *dynamoWrapper {
	var wrapper *dynamoWrapper = &dynamoWrapper{}
	lc.Append(
		fx.Hook{
			OnStart: func(ctx context.Context) (err error) {
				wrapper.LazyDynamoClient, err = connectToDynamo(ctx)
				return err
			},
		},
	)
        return wrapper
}

Everywhere you inject *dynamoWrapper during the constructors phase the client is not initialized, but once you start your Fx Application all the Start hooks will be initialized in order and once they do, *dynamoWrapper will have it's LazyDynamoClient properly initialized and ready to use.

In your example instead of injecting *dynamo.LazyDynamoClient inject the struct I proposed and make sure CreateDynamoWrapper is added before the Kafka constructor with the above LifeCycle OnStart Hook from your example.

[ttyerl]

Thanks!