@@ -12,7 +12,7 @@ import (
1212 "time"
1313
1414 "cloud.google.com/go/datastore"
15- "cloud.google.com/go/pubsub"
15+ "cloud.google.com/go/pubsub/v2 "
1616 "cloud.google.com/go/storage"
1717 "github.com/google/osv.dev/go/logger"
1818 "github.com/google/osv.dev/go/models"
@@ -118,7 +118,7 @@ func run(ctx context.Context, env *appEnv) error {
118118 go func () {
119119 defer resultsWg .Done ()
120120 for result := range resultsChan {
121- if handleResult (ctx , env .topic , result ) {
121+ if handleResult (ctx , env .publisher , result ) {
122122 newInvalids = append (newInvalids , result .id )
123123 }
124124 }
@@ -182,7 +182,7 @@ func run(ctx context.Context, env *appEnv) error {
182182// handleResult handles logging and sending pub/sub message to the recoverer.
183183// Returns true if a pub/sub message was sent to the recoverer,
184184// to indicate that we need to verify that the recoverer fixes the problem on the next run.
185- func handleResult (ctx context.Context , topic * pubsub.Topic , result checkRecordResult ) bool {
185+ func handleResult (ctx context.Context , publisher * pubsub.Publisher , result checkRecordResult ) bool {
186186 if result .err != nil {
187187 logger .Error ("failed to process record" , slog .String ("id" , result .id ), slog .Any ("err" , result .err ))
188188 }
@@ -191,7 +191,7 @@ func handleResult(ctx context.Context, topic *pubsub.Topic, result checkRecordRe
191191 Attributes : map [string ]string {"type" : "gcs_missing" , "id" : result .id },
192192 }
193193 logger .Info ("publishing gcs_missing message" , slog .String ("id" , result .id ))
194- _ , err := topic .Publish (ctx , & msg ).Get (ctx )
194+ _ , err := publisher .Publish (ctx , & msg ).Get (ctx )
195195 if err != nil {
196196 logger .Error ("failed publishing message" , slog .String ("id" , result .id ), slog .Any ("err" , err ))
197197 }
@@ -204,7 +204,7 @@ func handleResult(ctx context.Context, topic *pubsub.Topic, result checkRecordRe
204204type appEnv struct {
205205 bucket * storage.BucketHandle
206206 ds * datastore.Client
207- topic * pubsub.Topic
207+ publisher * pubsub.Publisher
208208 numWorkers int
209209}
210210
@@ -239,7 +239,7 @@ func setup(ctx context.Context) (*appEnv, error) {
239239 err = fmt .Errorf ("failed to create pubsub client: %w" , err )
240240 return nil , err
241241 }
242- topic := pubsubClient .Topic (pubsubTopic )
242+ publisher := pubsubClient .Publisher (pubsubTopic )
243243
244244 numWorkers := defaultNumWorkers
245245 if numWorkersStr , ok := os .LookupEnv ("NUM_WORKERS" ); ok {
@@ -253,7 +253,7 @@ func setup(ctx context.Context) (*appEnv, error) {
253253 return & appEnv {
254254 bucket : bucket ,
255255 ds : dsClient ,
256- topic : topic ,
256+ publisher : publisher ,
257257 numWorkers : numWorkers ,
258258 }, nil
259259}
0 commit comments