Skip to content

CROSSLINK-252 Add option to broadcast all event states #522

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JanisSaldabols wants to merge 1 commit into
base: main
Choose a base branch
from
+40 −5
Open

CROSSLINK-252 Add option to broadcast all event states #522

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions broker/client/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,12 +91,12 @@ func CreateIso18626ClientWithHttpClient(client *http.Client) Iso18626Client {

func (c *Iso18626Client) MessageRequester(ctx common.ExtendedContext, event events.Event) {
ctx = ctx.WithArgs(ctx.LoggerArgs().WithComponent(CLIENT_COMP))
_, _ = c.eventBus.ProcessTask(ctx, event, c.createAndSendSupplyingAgencyMessage)
_, _ = c.eventBus.ProcessTaskBroadcast(ctx, event, c.createAndSendSupplyingAgencyMessage)
Copy link
Copy Markdown
Contributor

@jakub-id jakub-id Apr 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@JanisSaldabols we need to test this somehow

}

func (c *Iso18626Client) MessageSupplier(ctx common.ExtendedContext, event events.Event) {
ctx = ctx.WithArgs(ctx.LoggerArgs().WithComponent(CLIENT_COMP))
_, _ = c.eventBus.ProcessTask(ctx, event, c.createAndSendRequestOrRequestingAgencyMessage)
_, _ = c.eventBus.ProcessTaskBroadcast(ctx, event, c.createAndSendRequestOrRequestingAgencyMessage)
}

func (c *Iso18626Client) getSkippedSupplierAndPeer(ctx common.ExtendedContext, illTransId, symbol string) (*ill_db.LocatedSupplier, *ill_db.Peer, error) {
Expand Down
39 changes: 37 additions & 2 deletions broker/events/eventbus.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,10 @@ type EventBus interface {
CreateNoticeWithParent(id string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, parentId *string) (string, error)
// Mark task for processing or fail if status is invalid (e.g already started)
BeginTask(eventId string) (Event, error)
BeginTaskBroadcast(eventId string) (Event, error)
// Mark task as completed or fail if status is invalid (e.g not started)
CompleteTask(eventId string, result *EventResult, status EventStatus) (Event, error)
CompleteTaskBroadcast(eventId string, result *EventResult, status EventStatus) (Event, error)
// Register handler for event (task or notice) created signal
HandleEventCreated(eventName EventName, f func(ctx common.ExtendedContext, event Event))
// Register handler for task started signal
Expand All @@ -45,6 +47,7 @@ type EventBus interface {
HandleTaskCompleted(eventName EventName, f func(ctx common.ExtendedContext, event Event))
// Execute task processing function within an automatic Begin/Complete block.
ProcessTask(ctx common.ExtendedContext, event Event, h func(common.ExtendedContext, Event) (EventStatus, *EventResult)) (Event, error)
ProcessTaskBroadcast(ctx common.ExtendedContext, event Event, h func(common.ExtendedContext, Event) (EventStatus, *EventResult)) (Event, error)
FindAncestor(descendant *Event, eventName EventName) *Event
GetLatestRequestEventByAction(ctx common.ExtendedContext, illTransId string, action string) (Event, error)
}
Expand Down Expand Up @@ -270,6 +273,14 @@ func (p *PostgresEventBus) createNotice(classId string, eventName EventName, dat
}

func (p *PostgresEventBus) BeginTask(eventId string) (Event, error) {
return p.beginTask(eventId, false)
}

func (p *PostgresEventBus) BeginTaskBroadcast(eventId string) (Event, error) {
return p.beginTask(eventId, true)
}

func (p *PostgresEventBus) beginTask(eventId string, broadcast bool) (Event, error) {
var event Event
err := p.repo.WithTxFunc(p.ctx, func(eventRepo EventRepo) error {
var err error
Expand All @@ -283,10 +294,15 @@ func (p *PostgresEventBus) BeginTask(eventId string) (Event, error) {
if event.EventStatus != EventStatusNew {
return fmt.Errorf("cannot begin task processing, event is not in state NEW but %s", event.EventStatus)
}
broadcastValue := event.Broadcast
if broadcast {
broadcastValue = broadcast
}
event, err = eventRepo.UpdateEventStatus(p.ctx, UpdateEventStatusParams{
ID: eventId,
EventStatus: EventStatusProcessing,
LastSignal: string(SignalTaskBegin),
Broadcast: broadcastValue,
})
if err != nil {
return err
Expand All @@ -298,6 +314,14 @@ func (p *PostgresEventBus) BeginTask(eventId string) (Event, error) {
}

func (p *PostgresEventBus) CompleteTask(eventId string, result *EventResult, status EventStatus) (Event, error) {
return p.completeTask(eventId, result, status, false)
}

func (p *PostgresEventBus) CompleteTaskBroadcast(eventId string, result *EventResult, status EventStatus) (Event, error) {
return p.completeTask(eventId, result, status, true)
}

func (p *PostgresEventBus) completeTask(eventId string, result *EventResult, status EventStatus, broadcast bool) (Event, error) {
var event Event
err := p.repo.WithTxFunc(p.ctx, func(eventRepo EventRepo) error {
var err error
Expand All @@ -316,6 +340,9 @@ func (p *PostgresEventBus) CompleteTask(eventId string, result *EventResult, sta
event.ResultData = *result
}
event.LastSignal = string(SignalTaskComplete)
if broadcast {
event.Broadcast = broadcast
}
event, err = eventRepo.SaveEvent(p.ctx, SaveEventParams(event))
if err != nil {
return err
Expand Down Expand Up @@ -357,16 +384,24 @@ func (p *PostgresEventBus) HandleTaskCompleted(eventName EventName, f func(ctx c
}

func (p *PostgresEventBus) ProcessTask(ctx common.ExtendedContext, event Event, h func(common.ExtendedContext, Event) (EventStatus, *EventResult)) (Event, error) {
return p.processTask(ctx, event, h, p.BeginTask, p.CompleteTask)
}

func (p *PostgresEventBus) ProcessTaskBroadcast(ctx common.ExtendedContext, event Event, h func(common.ExtendedContext, Event) (EventStatus, *EventResult)) (Event, error) {
return p.processTask(ctx, event, h, p.BeginTaskBroadcast, p.CompleteTaskBroadcast)
}
Comment on lines +390 to +392

func (p *PostgresEventBus) processTask(ctx common.ExtendedContext, event Event, h func(common.ExtendedContext, Event) (EventStatus, *EventResult), b func(eventId string) (Event, error), c func(eventId string, result *EventResult, status EventStatus) (Event, error)) (Event, error) {
inEvent := &event
event, err := p.BeginTask(event.ID)
event, err := b(event.ID)
Copy link
Copy Markdown
Contributor

@jakub-id jakub-id Apr 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we call these handlers something more readable like begin and complete?

if err != nil {
p.getEventContext(inEvent).Logger().Warn("failed to start processing TASK event", "error", err, "eventName", inEvent.EventName)
return event, err
}

status, result := h(ctx, event)

event, err = p.CompleteTask(event.ID, result, status)
event, err = c(event.ID, result, status)
if err != nil {
p.getEventContext(inEvent).Logger().Warn("failed to complete processing TASK event", "error", err, "eventName", inEvent.EventName)
return event, err
Expand Down
2 changes: 1 addition & 1 deletion broker/sqlc/event_query.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ DELETE FROM event
WHERE ill_transaction_id = $1;

-- name: UpdateEventStatus :one
Copy link
Copy Markdown
Contributor

@jakub-id jakub-id Apr 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe rename this to something like UpdateEventLifecycle

UPDATE event SET last_signal = $3, event_status = $2
UPDATE event SET last_signal = $3, event_status = $2, broadcast = $4
WHERE id = $1
RETURNING sqlc.embed(event);

Expand Down
Loading