Skip to content

Commit

Permalink
rename wgCtx/wgErr
Browse files Browse the repository at this point in the history
  • Loading branch information
@serprex
serprex committed May 7, 2024
1 parent 2b338d2 commit 3badb8d
Show file tree
Hide file tree
Showing 4 changed files with 42 additions and 42 deletions.
20 changes: 10 additions & 10 deletions flow/connectors/kafka/kafka.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -169,11 +169,11 @@ func (c *KafkaConnector) createPool(
ctx context.Context,
script string,
flowJobName string,
wgErr func(error),
queueErr func(error),
) (*utils.LPool[[]*kgo.Record], error) {
produceCb := func(_ *kgo.Record, err error) {
if err != nil {
wgErr(err)
queueErr(err)
}
}

Expand Down Expand Up @@ -205,8 +205,8 @@ func (c *KafkaConnector) SyncRecords(ctx context.Context, req *model.SyncRecords
numRecords := atomic.Int64{}
tableNameRowsMapping := utils.InitialiseTableRowsMap(req.TableMappings)

wgCtx, wgErr := context.WithCancelCause(ctx)
pool, err := c.createPool(wgCtx, req.Script, req.FlowJobName, wgErr)
queueCtx, queueErr := context.WithCancelCause(ctx)
pool, err := c.createPool(queueCtx, req.Script, req.FlowJobName, queueErr)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -255,15 +255,15 @@ Loop:
lfn := ls.Env.RawGetString("onRecord")
fn, ok := lfn.(*lua.LFunction)
if !ok {
wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
return nil
}

ls.Push(fn)
ls.Push(pua.LuaRecord.New(ls, record))
err := ls.PCall(1, -1, nil)
if err != nil {
wgErr(fmt.Errorf("script failed: %w", err))
queueErr(fmt.Errorf("script failed: %w", err))
return nil
}

Expand All @@ -272,7 +272,7 @@ Loop:
for i := range args {
kr, err := lvalueToKafkaRecord(ls, ls.Get(i-args))
if err != nil {
wgErr(err)
queueErr(err)
return nil
}
if kr != nil {
Expand All @@ -289,16 +289,16 @@ Loop:
return results
})

case <-wgCtx.Done():
case <-queueCtx.Done():
break Loop
}
}

close(flushLoopDone)
if err := pool.Wait(wgCtx); err != nil {
if err := pool.Wait(queueCtx); err != nil {
return nil, err
}
if err := c.client.Flush(wgCtx); err != nil {
if err := c.client.Flush(queueCtx); err != nil {
return nil, fmt.Errorf("[kafka] final flush error: %w", err)
}

Expand Down
16 changes: 8 additions & 8 deletions flow/connectors/kafka/qrep.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ func (c *KafkaConnector) SyncQRepRecords(
numRecords := atomic.Int64{}
schema := stream.Schema()

wgCtx, wgErr := context.WithCancelCause(ctx)
pool, err := c.createPool(wgCtx, config.Script, config.FlowJobName, wgErr)
queueCtx, queueErr := context.WithCancelCause(ctx)
pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, queueErr)
if err != nil {
return 0, err
}
Expand Down Expand Up @@ -60,15 +60,15 @@ Loop:
lfn := ls.Env.RawGetString("onRecord")
fn, ok := lfn.(*lua.LFunction)
if !ok {
wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
return nil
}

ls.Push(fn)
ls.Push(pua.LuaRecord.New(ls, record))
err := ls.PCall(1, -1, nil)
if err != nil {
wgErr(fmt.Errorf("script failed: %w", err))
queueErr(fmt.Errorf("script failed: %w", err))
return nil
}

Expand All @@ -77,7 +77,7 @@ Loop:
for i := range args {
kr, err := lvalueToKafkaRecord(ls, ls.Get(i-args))
if err != nil {
wgErr(err)
queueErr(err)
return nil
}
if kr != nil {
Expand All @@ -92,15 +92,15 @@ Loop:
return results
})

case <-wgCtx.Done():
case <-queueCtx.Done():
break Loop
}
}

if err := pool.Wait(wgCtx); err != nil {
if err := pool.Wait(queueCtx); err != nil {
return 0, err
}
if err := c.client.Flush(wgCtx); err != nil {
if err := c.client.Flush(queueCtx); err != nil {
return 0, fmt.Errorf("[kafka] final flush error: %w", err)
}

Expand Down
26 changes: 13 additions & 13 deletions flow/connectors/pubsub/pubsub.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func (c *PubSubConnector) createPool(
flowJobName string,
topiccache *topicCache,
publish chan<- *pubsub.PublishResult,
wgErr func(error),
queueErr func(error),
) (*utils.LPool[[]PubSubMessage], error) {
return utils.LuaPool(func() (*lua.LState, error) {
ls, err := utils.LoadScript(ctx, script, func(ls *lua.LState) int {
Expand Down Expand Up @@ -163,7 +163,7 @@ func (c *PubSubConnector) createPool(
return topicClient, nil
})
if err != nil {
wgErr(err)
queueErr(err)
return
}

Expand Down Expand Up @@ -228,8 +228,8 @@ func (c *PubSubConnector) SyncRecords(ctx context.Context, req *model.SyncRecord
publish := make(chan *pubsub.PublishResult, 32)
waitChan := make(chan struct{})

wgCtx, wgErr := context.WithCancelCause(ctx)
pool, err := c.createPool(wgCtx, req.Script, req.FlowJobName, &topiccache, publish, wgErr)
queueCtx, queueErr := context.WithCancelCause(ctx)
pool, err := c.createPool(queueCtx, req.Script, req.FlowJobName, &topiccache, publish, queueErr)
if err != nil {
return nil, err
}
Expand All @@ -238,7 +238,7 @@ func (c *PubSubConnector) SyncRecords(ctx context.Context, req *model.SyncRecord
go func() {
for curpub := range publish {
if _, err := curpub.Get(ctx); err != nil {
wgErr(err)
queueErr(err)
break
}
}
Expand Down Expand Up @@ -285,15 +285,15 @@ Loop:
lfn := ls.Env.RawGetString("onRecord")
fn, ok := lfn.(*lua.LFunction)
if !ok {
wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
return nil
}

ls.Push(fn)
ls.Push(pua.LuaRecord.New(ls, record))
err := ls.PCall(1, -1, nil)
if err != nil {
wgErr(fmt.Errorf("script failed: %w", err))
queueErr(fmt.Errorf("script failed: %w", err))
return nil
}

Expand All @@ -302,7 +302,7 @@ Loop:
for i := range args {
msg, err := lvalueToPubSubMessage(ls, ls.Get(i-args))
if err != nil {
wgErr(err)
queueErr(err)
return nil
}
if msg.Message != nil {
Expand All @@ -319,20 +319,20 @@ Loop:
return results
})

case <-wgCtx.Done():
case <-queueCtx.Done():
break Loop
}
}

close(flushLoopDone)
if err := pool.Wait(wgCtx); err != nil {
if err := pool.Wait(queueCtx); err != nil {
return nil, err
}
close(publish)
topiccache.Stop(wgCtx)
topiccache.Stop(queueCtx)
select {
case <-wgCtx.Done():
return nil, wgCtx.Err()
case <-queueCtx.Done():
return nil, queueCtx.Err()
case <-waitChan:
}

Expand Down
22 changes: 11 additions & 11 deletions flow/connectors/pubsub/qrep.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@ func (c *PubSubConnector) SyncQRepRecords(
publish := make(chan *pubsub.PublishResult, 32)
waitChan := make(chan struct{})

wgCtx, wgErr := context.WithCancelCause(ctx)
pool, err := c.createPool(wgCtx, config.Script, config.FlowJobName, &topiccache, publish, wgErr)
queueCtx, queueErr := context.WithCancelCause(ctx)
pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, &topiccache, publish, queueErr)
if err != nil {
return 0, err
}
Expand All @@ -41,7 +41,7 @@ func (c *PubSubConnector) SyncQRepRecords(
go func() {
for curpub := range publish {
if _, err := curpub.Get(ctx); err != nil {
wgErr(err)
queueErr(err)
break
}
}
Expand Down Expand Up @@ -73,15 +73,15 @@ Loop:
lfn := ls.Env.RawGetString("onRecord")
fn, ok := lfn.(*lua.LFunction)
if !ok {
wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
return nil
}

ls.Push(fn)
ls.Push(pua.LuaRecord.New(ls, record))
err := ls.PCall(1, -1, nil)
if err != nil {
wgErr(fmt.Errorf("script failed: %w", err))
queueErr(fmt.Errorf("script failed: %w", err))
return nil
}

Expand All @@ -90,7 +90,7 @@ Loop:
for i := range args {
msg, err := lvalueToPubSubMessage(ls, ls.Get(i-args))
if err != nil {
wgErr(err)
queueErr(err)
return nil
}
if msg.Message != nil {
Expand All @@ -105,19 +105,19 @@ Loop:
return results
})

case <-wgCtx.Done():
case <-queueCtx.Done():
break Loop
}
}

if err := pool.Wait(wgCtx); err != nil {
if err := pool.Wait(queueCtx); err != nil {
return 0, err
}
close(publish)
topiccache.Stop(wgCtx)
topiccache.Stop(queueCtx)
select {
case <-wgCtx.Done():
return 0, wgCtx.Err()
case <-queueCtx.Done():
return 0, queueCtx.Err()
case <-waitChan:
}

Expand Down

0 comments on commit 3badb8d

Please sign in to comment.