forked from TrueCloudLab/frostfs-node
64 lines
1.4 KiB
Go
64 lines
1.4 KiB
Go
|
package patchsvc
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/ecdsa"
|
||
|
|
"io"
|
||
|
|
|
||
|
|
getsvc "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/get"
|
||
|
|
objectUtil "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
|
||
|
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||
|
|
patcherSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/patcher"
|
||
|
|
)
|
||
|
|
|
||
|
|
func (p *pipeChunkWriter) WriteChunk(_ context.Context, chunk []byte) error {
|
||
|
|
_, err := p.wr.Write(chunk)
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
type rangeProvider struct {
|
||
|
|
getSvc *getsvc.Service
|
||
|
|
|
||
|
|
addr oid.Address
|
||
|
|
|
||
|
|
commonPrm *objectUtil.CommonPrm
|
||
|
|
|
||
|
|
localNodeKey *ecdsa.PrivateKey
|
||
|
|
}
|
||
|
|
|
||
|
|
var _ patcherSDK.RangeProvider = (*rangeProvider)(nil)
|
||
|
|
|
||
|
|
func (r *rangeProvider) GetRange(ctx context.Context, rng *objectSDK.Range) io.Reader {
|
||
|
|
pipeReader, pipeWriter := io.Pipe()
|
||
|
|
|
||
|
|
var rngPrm getsvc.RangePrm
|
||
|
|
rngPrm.SetSignerKey(r.localNodeKey)
|
||
|
|
rngPrm.SetCommonParameters(r.commonPrm)
|
||
|
|
|
||
|
|
rngPrm.WithAddress(r.addr)
|
||
|
|
rngPrm.SetChunkWriter(&pipeChunkWriter{
|
||
|
|
wr: pipeWriter,
|
||
|
|
})
|
||
|
|
rngPrm.SetRange(rng)
|
||
|
|
|
||
|
|
getRangeErr := make(chan error)
|
||
|
|
|
||
|
|
go func() {
|
||
|
|
defer pipeWriter.Close()
|
||
|
|
|
||
|
|
select {
|
||
|
|
case <-ctx.Done():
|
||
|
|
pipeWriter.CloseWithError(ctx.Err())
|
||
|
|
case err := <-getRangeErr:
|
||
|
|
pipeWriter.CloseWithError(err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
go func() {
|
||
|
|
getRangeErr <- r.getSvc.GetRange(ctx, rngPrm)
|
||
|
|
}()
|
||
|
|
|
||
|
|
return pipeReader
|
||
|
|
}
|