diff options
| author | Chris Lu <chris.lu@gmail.com> | 2020-07-19 17:59:43 -0700 |
|---|---|---|
| committer | Chris Lu <chris.lu@gmail.com> | 2020-07-19 17:59:43 -0700 |
| commit | 97d97f35287a88de392a1a422b3533339d923ae2 (patch) | |
| tree | f69b1c2aea74ffc10a9b1ef031572a0f34decfce /weed/filesys/wfs_write.go | |
| parent | f90d2c93c9e34997a8e76aeefb438ec06b1cd093 (diff) | |
| download | seaweedfs-97d97f35287a88de392a1a422b3533339d923ae2.tar.xz seaweedfs-97d97f35287a88de392a1a422b3533339d923ae2.zip | |
go code can read and write chunk manifest
Diffstat (limited to 'weed/filesys/wfs_write.go')
| -rw-r--r-- | weed/filesys/wfs_write.go | 66 |
1 files changed, 66 insertions, 0 deletions
diff --git a/weed/filesys/wfs_write.go b/weed/filesys/wfs_write.go new file mode 100644 index 000000000..786d0b42a --- /dev/null +++ b/weed/filesys/wfs_write.go @@ -0,0 +1,66 @@ +package filesys + +import ( + "context" + "fmt" + "io" + + "github.com/chrislusf/seaweedfs/weed/filer2" + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/operation" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/security" +) + +func (wfs *WFS) saveDataAsChunk(dir string) filer2.SaveDataAsChunkFunctionType { + + return func(reader io.Reader, filename string, offset int64) (chunk *filer_pb.FileChunk, collection, replication string, err error) { + var fileId, host string + var auth security.EncodedJwt + + if err := wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { + + request := &filer_pb.AssignVolumeRequest{ + Count: 1, + Replication: wfs.option.Replication, + Collection: wfs.option.Collection, + TtlSec: wfs.option.TtlSec, + DataCenter: wfs.option.DataCenter, + ParentPath: dir, + } + + resp, err := client.AssignVolume(context.Background(), request) + if err != nil { + glog.V(0).Infof("assign volume failure %v: %v", request, err) + return err + } + if resp.Error != "" { + return fmt.Errorf("assign volume failure %v: %v", request, resp.Error) + } + + fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth) + host = wfs.AdjustedUrl(host) + collection, replication = resp.Collection, resp.Replication + + return nil + }); err != nil { + return nil, "", "", fmt.Errorf("filerGrpcAddress assign volume: %v", err) + } + + fileUrl := fmt.Sprintf("http://%s/%s", host, fileId) + uploadResult, err, data := operation.Upload(fileUrl, filename, wfs.option.Cipher, reader, false, "", nil, auth) + if err != nil { + glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err) + return nil, "", "", fmt.Errorf("upload data: %v", err) + } + if uploadResult.Error != "" { + glog.V(0).Infof("upload failure %v to %s: %v", filename, fileUrl, err) + return nil, "", "", fmt.Errorf("upload result: %v", uploadResult.Error) + } + + wfs.chunkCache.SetChunk(fileId, data) + + chunk = uploadResult.ToPbFileChunk(fileId, offset) + return chunk, "", "", nil + } +} |
