// createOrUpdateEventSource will do:
// 1. Generate a dapr component specification for the EventBus associated with the EventSource (if spec.eventBus is set)
//    and create the dapr component (will check if it needs to be updated)
//    and set the EventSourceConfig.EventBusComponent, EventSourceConfig.EventBusTopic, EventSourceConfig.EventBusSpecEncode.
// 2. Generate a dapr component specification for the Sink set in EventSource (if spec.sink is set)
//    and create the dapr component (will check if it needs to be updated)
//    and set the EventSourceConfig.SinkComponent, EventSourceConfig.SinkSpecEncode.
// 3. Generate dapr component specifications for the event sources
//    and create the dapr components (will check if they need to be updated)
//    and set the EventSourceConfig.EventSourceComponent, EventSourceConfig.EventSourceTopic, EventSourceConfig.EventSourceSpecEncode.
// 4. Generate EventSourceConfig and convert it to a base64-encoded string.
// 5. Create an EventSource workload for each event source (will check if they need to be updated)
//    and pass in the EventSourceConfig as an environment variable.
createOrUpdateEventSource will do:
1. Generate a dapr component specification for the EventBus associated with the EventSource (if spec.eventBus is set)
and create the dapr component (will check if it needs to be updated)
and set the EventSourceConfig.EventBusComponent, EventSourceConfig.EventBusTopic, EventSourceConfig.EventBusSpecEncode.
2. Generate a dapr component specification for the Sink set in EventSource (if spec.sink is set)
and create the dapr component (will check if it needs to be updated)
and set the EventSourceConfig.SinkComponent, EventSourceConfig.SinkSpecEncode.
3. Generate dapr component specifications for the event sources
and create the dapr components (will check if they need to be updated)
and set the EventSourceConfig.EventSourceComponent, EventSourceConfig.EventSourceTopic, EventSourceConfig.EventSourceSpecEncode.
4. Generate EventSourceConfig and convert it to a base64-encoded string.
5. Create an EventSource workload for each event source (will check if they need to be updated)
and pass in the EventSourceConfig as an environment variable.
// createOrUpdateEventSource will do : // 1 . Generate a dapr component specification for the EventBus associated with the EventSource ( if spec . eventBus is set ) // and create the dapr component ( will check if it needs to be updated ) // and set the EventSourceConfig . EventBusComponent EventSourceConfig . EventBusTopic EventSourceConfig . EventBusSpecEncode . // 2 . Generate a dapr component specification for the Sink set in EventSource ( if spec . sink is set ) // and create the dapr component ( will check if it needs to be updated ) // and set the EventSourceConfig . SinkComponent EventSourceConfig . SinkSpecEncode . // 3 . Generate dapr component specifications for the event sources // and create the dapr components ( will check if they need to be updated ) // and set the EventSourceConfig . EventSourceComponent EventSourceConfig . EventSourceTopic EventSourceConfig . EventSourceSpecEncode . // 4 . Generate EventSourceConfig and convert it to a base64 - encoded string . // 5 . Create an EventSource workload for each event source ( will check if they need to be updated ) // and pass in the EventSourceConfig as an environment variable .
func (r *EventSourceReconciler) createOrUpdateEventSource(ctx context.Context, log logr.Logger, eventSource *ofevent.EventSource) error {
	defer eventSource.SaveStatus(ctx, log, r.Client)
	log = r.Log.WithName("createOrUpdateEventSource")
	eventSource.AddCondition(*ofevent.CreateCondition(
		ofevent.Pending, metav1.ConditionUnknown, ofevent.PendingCreation,
	).SetMessage("Identified EventSource creation signal"))
	r.Function = InitFunction(eventSourceHandlerImage)
	if eventSource.Spec.EventBus == "" && eventSource.Spec.Sink == nil {
		err := errors.New("must set spec.eventBus or spec.sink")
		condition := ofevent.CreateCondition(
			ofevent.Error, metav1.ConditionFalse, ofevent.ErrorConfiguration,
		).SetMessage(err.Error())
		eventSource.AddCondition(*condition)
		log.Error(err, "Failed to find output configuration (eventBus or sink).",
			"namespace", eventSource.Namespace, "name", eventSource.Name)
		return err
	}
	if eventSource.Spec.EventBus != "" {
		if err := r.handleEventBus(ctx, log, eventSource); err != nil {
			return err
		}
	}
	if eventSource.Spec.Sink != nil {
		if err := r.handleSink(ctx, log, eventSource); err != nil {
			return err
		}
	}
	if err := r.handleEventSource(ctx, log, eventSource); err != nil {
		return err
	}
	if _, err := ctrl.CreateOrUpdate(ctx, r.Client, eventSource, r.mutateEventSource(eventSource)); err != nil {
		condition := ofevent.CreateCondition(
			ofevent.Error, metav1.ConditionFalse, ofevent.ErrorCreatingEventSource,
		).SetMessage(err.Error())
		eventSource.AddCondition(*condition)
		log.Error(err, "Failed to create or update EventSource",
			"namespace", eventSource.Namespace, "name", eventSource.Name)
		return err
	}
	condition := ofevent.CreateCondition(
		ofevent.Ready, metav1.ConditionTrue, ofevent.EventSourceIsReady,
	).SetMessage("EventSource is ready.")
	eventSource.AddCondition(*condition)
	eventSource.SaveStatus(ctx, log, r.Client)
	log.Info("EventSource reconcile success.",
		"namespace", eventSource.Namespace, "name", eventSource.Name)
	return nil
}
func ( r * EventSourceReconciler ) createOrUpdateEventSource ( ctx context . Context , log logr . Logger , eventSource * ofevent . EventSource ) error { defer eventSource . SaveStatus ( ctx , log , r . Client ) log = r . Log . WithName ( " " ) eventSource . AddCondition ( * ofevent . CreateCondition ( ofevent . Pending , metav1 . ConditionUnknown , ofevent . PendingCreation , ) . SetMessage ( " " ) ) r . Function = InitFunction ( eventSourceHandlerImage ) if eventSource . Spec . EventBus == " " && eventSource . Spec . Sink == nil { err := errors . New ( " " ) condition := ofevent . CreateCondition ( ofevent . Error , metav1 . ConditionFalse , ofevent . ErrorConfiguration , ) . SetMessage ( err . Error ( ) ) eventSource . AddCondition ( * condition ) log . Error ( err , " " , " " , eventSource . Namespace , " " , eventSource . Name ) return err } if eventSource . Spec . EventBus != " " { if err := r . handleEventBus ( ctx , log , eventSource ) ; err != nil { return err } } if eventSource . Spec . Sink != nil { if err := r . handleSink ( ctx , log , eventSource ) ; err != nil { return err } } if err := r . handleEventSource ( ctx , log , eventSource ) ; err != nil { return err } if _ , err := ctrl . CreateOrUpdate ( ctx , r . Client , eventSource , r . mutateEventSource ( eventSource ) ) ; err != nil { condition := ofevent . CreateCondition ( ofevent . Error , metav1 . ConditionFalse , ofevent . ErrorCreatingEventSource , ) . SetMessage ( err . Error ( ) ) eventSource . AddCondition ( * condition ) log . Error ( err , " " , " " , eventSource . Namespace , " " , eventSource . Name ) return err } condition := ofevent . CreateCondition ( ofevent . Ready , metav1 . ConditionTrue , ofevent . EventSourceIsReady , ) . SetMessage ( " " ) eventSource . AddCondition ( * condition ) eventSource . SaveStatus ( ctx , log , r . Client ) log . Info ( " " , " " , eventSource . Namespace , " " , eventSource . Name ) return nil }
createOrUpdateEventSource will do:
1.
createOrUpdateEventSource will do : 1 .
// Generate the eventsource function instance with image eventSourceHandlerImage. // Handle EventBus(ClusterEventBus) reconcile. // Handle Sink reconcile. // Handle EventSource reconcile.