This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch lostluck-protosuffix
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/lostluck-protosuffix by this
push:
new f449dca Update stage.go
f449dca is described below
commit f449dca29e374e324c1bdf1325b6091dbf65c96c
Author: Robert Burke <[email protected]>
AuthorDate: Tue Apr 7 22:14:15 2020 -0700
Update stage.go
Replace generic pb shortname with jobpb
---
sdks/go/pkg/beam/artifact/stage.go | 32 ++++++++++++++++----------------
1 file changed, 16 insertions(+), 16 deletions(-)
diff --git a/sdks/go/pkg/beam/artifact/stage.go
b/sdks/go/pkg/beam/artifact/stage.go
index 87cf557..c8a5f9b 100644
--- a/sdks/go/pkg/beam/artifact/stage.go
+++ b/sdks/go/pkg/beam/artifact/stage.go
@@ -30,15 +30,15 @@ import (
"time"
"github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
- pb "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
+ jobpb "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
"github.com/apache/beam/sdks/go/pkg/beam/util/errorx"
)
// Commit commits a manifest with the given staged artifacts. It returns the
// staging token, if successful.
-func Commit(ctx context.Context, client pb.LegacyArtifactStagingServiceClient,
artifacts []*pb.ArtifactMetadata, st string) (string, error) {
- req := &pb.CommitManifestRequest{
- Manifest: &pb.Manifest{
+func Commit(ctx context.Context, client
jobpb.LegacyArtifactStagingServiceClient, artifacts []*jobpb.ArtifactMetadata,
st string) (string, error) {
+ req := &jobpb.CommitManifestRequest{
+ Manifest: &jobpb.Manifest{
Artifact: artifacts,
},
StagingSessionToken: st,
@@ -51,7 +51,7 @@ func Commit(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, a
}
// StageDir stages a local directory with relative path keys. Convenience
wrapper.
-func StageDir(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, src string, st string)
([]*pb.ArtifactMetadata, error) {
+func StageDir(ctx context.Context, client
jobpb.LegacyArtifactStagingServiceClient, src string, st string)
([]*jobpb.ArtifactMetadata, error) {
list, err := scan(src)
if err != nil || len(list) == 0 {
return nil, err
@@ -62,7 +62,7 @@ func StageDir(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient,
// MultiStage stages a set of local files with the given keys. It returns
// the full artifact metadate. It retries each artifact a few times.
// Convenience wrapper.
-func MultiStage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, cpus int, list []KeyedFile, st string)
([]*pb.ArtifactMetadata, error) {
+func MultiStage(ctx context.Context, client
jobpb.LegacyArtifactStagingServiceClient, cpus int, list []KeyedFile, st
string) ([]*jobpb.ArtifactMetadata, error) {
if cpus < 1 {
cpus = 1
}
@@ -77,7 +77,7 @@ func MultiStage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClien
close(q)
var permErr errorx.GuardedError
- ret := make(chan *pb.ArtifactMetadata, len(list))
+ ret := make(chan *jobpb.ArtifactMetadata, len(list))
var wg sync.WaitGroup
for i := 0; i < cpus; i++ {
@@ -119,7 +119,7 @@ func MultiStage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClien
// Stage stages a local file as an artifact with the given key. It computes
// the SHA256 and returns the full artifact metadata.
-func Stage(ctx context.Context, client pb.LegacyArtifactStagingServiceClient,
key, filename, st string) (*pb.ArtifactMetadata, error) {
+func Stage(ctx context.Context, client
jobpb.LegacyArtifactStagingServiceClient, key, filename, st string)
(*jobpb.ArtifactMetadata, error) {
stat, err := os.Stat(filename)
if err != nil {
return nil, err
@@ -128,12 +128,12 @@ func Stage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, ke
if err != nil {
return nil, err
}
- md := &pb.ArtifactMetadata{
+ md := &jobpb.ArtifactMetadata{
Name: key,
Permissions: uint32(stat.Mode()),
Sha256: hash,
}
- pmd := &pb.PutArtifactMetadata{
+ pmd := &jobpb.PutArtifactMetadata{
Metadata: md,
StagingSessionToken: st,
}
@@ -149,8 +149,8 @@ func Stage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, ke
return nil, err
}
- header := &pb.PutArtifactRequest{
- Content: &pb.PutArtifactRequest_Metadata{
+ header := &jobpb.PutArtifactRequest{
+ Content: &jobpb.PutArtifactRequest_Metadata{
Metadata: pmd,
},
}
@@ -172,7 +172,7 @@ func Stage(ctx context.Context, client
pb.LegacyArtifactStagingServiceClient, ke
return md, nil
}
-func stageChunks(stream pb.LegacyArtifactStagingService_PutArtifactClient, r
io.Reader) (string, error) {
+func stageChunks(stream jobpb.LegacyArtifactStagingService_PutArtifactClient,
r io.Reader) (string, error) {
sha256W := sha256.New()
data := make([]byte, 1<<20)
for {
@@ -182,9 +182,9 @@ func stageChunks(stream
pb.LegacyArtifactStagingService_PutArtifactClient, r io.
panic(err) // cannot fail
}
- chunk := &pb.PutArtifactRequest{
- Content: &pb.PutArtifactRequest_Data{
- Data: &pb.ArtifactChunk{
+ chunk := &jobpb.PutArtifactRequest{
+ Content: &jobpb.PutArtifactRequest_Data{
+ Data: &jobpb.ArtifactChunk{
Data: data[:n],
},
},
