refactor: finality check, KV client

cmd/upload.go

@@ -26,8 +26,8 @@ var (
 
 		node []string
 
-		force    bool
-		taskSize uint
+		finalityRequired bool
+		taskSize         uint
 	}
 
 	uploadCmd = &cobra.Command{
@@ -52,7 +52,7 @@ func init() {
 	uploadCmd.Flags().StringSliceVar(&uploadArgs.node, "node", []string{}, "ZeroGStorage storage node URL")
 	uploadCmd.MarkFlagRequired("node")
 
-	uploadCmd.Flags().BoolVar(&uploadArgs.force, "force", false, "Force to upload file even already exists")
+	uploadCmd.Flags().BoolVar(&uploadArgs.finalityRequired, "finality-required", false, "Wait for file finality on nodes to upload")
 	uploadCmd.Flags().UintVar(&uploadArgs.taskSize, "task-size", 10, "Number of segments to upload in single rpc request")
 
 	rootCmd.AddCommand(uploadCmd)
@@ -76,9 +76,9 @@ func upload(*cobra.Command, []string) {
 		logrus.WithError(err).Fatal("Failed to initialize uploader")
 	}
 	opt := transfer.UploadOption{
-		Tags:     hexutil.MustDecode(uploadArgs.tags),
-		Force:    uploadArgs.force,
-		TaskSize: uploadArgs.taskSize,
+		Tags:             hexutil.MustDecode(uploadArgs.tags),
+		FinalityRequired: uploadArgs.finalityRequired,
+		TaskSize:         uploadArgs.taskSize,
 	}
 
 	file, err := core.Open(uploadArgs.file)

example/kv_put/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"math"
 
 	"github.com/0glabs/0g-storage-client/common/blockchain"
 	"github.com/0glabs/0g-storage-client/contract"
@@ -25,13 +26,12 @@ func main() {
 	blockchainClient := blockchain.MustNewWeb3(BlockchainClientAddr, PrivKey)
 	defer blockchainClient.Close()
 	blockchain.CustomGasLimit = 1000000
-	zgs, err := contract.NewFlowContract(common.HexToAddress(FlowContractAddr), blockchainClient)
+	flow, err := contract.NewFlowContract(common.HexToAddress(FlowContractAddr), blockchainClient)
 	if err != nil {
 		fmt.Println(err)
 		return
 	}
-	kvClient := kv.NewClient(zgsClient, zgs)
-	batcher := kvClient.Batcher()
+	batcher := kv.NewBatcher(math.MaxUint64, []*node.Client{zgsClient}, flow)
 	batcher.Set(common.HexToHash("0x000000000000000000000000000000000000000000000000000000000000f2bd"),
 		[]byte("TESTKEY0"),
 		[]byte{69, 70, 71, 72, 73},

kv/batcher.go

@@ -0,0 +1,63 @@
+package kv
+
+import (
+	"context"
+
+	zg_common "github.com/0glabs/0g-storage-client/common"
+	"github.com/0glabs/0g-storage-client/contract"
+	"github.com/0glabs/0g-storage-client/core"
+	"github.com/0glabs/0g-storage-client/node"
+	"github.com/0glabs/0g-storage-client/transfer"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+type Batcher struct {
+	*StreamDataBuilder
+	clients []*node.Client
+	flow    *contract.FlowContract
+	logger  *logrus.Logger
+}
+
+func NewBatcher(version uint64, clients []*node.Client, flow *contract.FlowContract, opts ...zg_common.LogOption) *Batcher {
+	return &Batcher{
+		StreamDataBuilder: NewStreamDataBuilder(version),
+		clients:           clients,
+		flow:              flow,
+		logger:            zg_common.NewLogger(opts...),
+	}
+}
+
+// Exec submit the kv operations to ZeroGStorage network in batch.
+//
+// Note, this is a time consuming operation, e.g. several seconds or even longer.
+// When it comes to a time sentitive context, it should be executed in a separate go-routine.
+func (b *Batcher) Exec(ctx context.Context) error {
+	// build stream data
+	streamData, err := b.Build()
+	if err != nil {
+		return errors.WithMessage(err, "Failed to build stream data")
+	}
+
+	encoded, err := streamData.Encode()
+	if err != nil {
+		return errors.WithMessage(err, "Failed to encode data")
+	}
+	data, err := core.NewDataInMemory(encoded)
+	if err != nil {
+		return err
+	}
+
+	// upload file
+	uploader, err := transfer.NewUploader(b.flow, b.clients, zg_common.LogOption{Logger: b.logger})
+	if err != nil {
+		return err
+	}
+	opt := transfer.UploadOption{
+		Tags: b.BuildTags(),
+	}
+	if err = uploader.Upload(ctx, data, opt); err != nil {
+		return errors.WithMessagef(err, "Failed to upload data")
+	}
+	return nil
+}

kv/client.go

@@ -4,14 +4,9 @@ import (
 	"context"
 	"math"
 
-	zg_common "github.com/0glabs/0g-storage-client/common"
 	"github.com/0glabs/0g-storage-client/contract"
-	"github.com/0glabs/0g-storage-client/core"
 	"github.com/0glabs/0g-storage-client/node"
-	"github.com/0glabs/0g-storage-client/transfer"
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 )
 
 // Client is used for users to communicate with server for kv operations.
@@ -20,10 +15,7 @@ type Client struct {
 	flow *contract.FlowContract
 }
 
-// NewClient creates a new client for kv operations.
-//
-// Generally, you could refer to the `upload` function in `cmd/upload.go` file
-// for how to create storage node client and flow contract client.
+// NewClient creates a new client for kv queries.
 func NewClient(node *node.Client, flow *contract.FlowContract) *Client {
 	return &Client{
 		node: node,
@@ -124,57 +116,3 @@ func (c *Client) IsWriterOfKey(ctx context.Context, account common.Address, stre
 func (c *Client) IsWriterOfStream(ctx context.Context, account common.Address, streamId common.Hash, version ...uint64) (isWriter bool, err error) {
 	return c.node.KV().IsWriterOfStream(ctx, account, streamId, version...)
 }
-
-// Batcher returns a Batcher instance for kv operations in batch.
-func (c *Client) Batcher(opts ...zg_common.LogOption) *Batcher {
-	return newBatcher(math.MaxUint64, c, opts...)
-}
-
-type Batcher struct {
-	*StreamDataBuilder
-	client *Client
-	logger *logrus.Logger
-}
-
-func newBatcher(version uint64, client *Client, opts ...zg_common.LogOption) *Batcher {
-	return &Batcher{
-		StreamDataBuilder: NewStreamDataBuilder(version),
-		client:            client,
-		logger:            zg_common.NewLogger(opts...),
-	}
-}
-
-// Exec submit the kv operations to ZeroGStorage network in batch.
-//
-// Note, this is a time consuming operation, e.g. several seconds or even longer.
-// When it comes to a time sentitive context, it should be executed in a separate go-routine.
-func (b *Batcher) Exec(ctx context.Context) error {
-	// build stream data
-	streamData, err := b.Build()
-	if err != nil {
-		return errors.WithMessage(err, "Failed to build stream data")
-	}
-
-	encoded, err := streamData.Encode()
-	if err != nil {
-		return errors.WithMessage(err, "Failed to encode data")
-	}
-	data, err := core.NewDataInMemory(encoded)
-	if err != nil {
-		return err
-	}
-
-	// upload file
-	uploader, err := transfer.NewUploader(b.client.flow, []*node.Client{b.client.node}, zg_common.LogOption{Logger: b.logger})
-	if err != nil {
-		return err
-	}
-	opt := transfer.UploadOption{
-		Tags:  b.BuildTags(),
-		Force: true,
-	}
-	if err = uploader.Upload(ctx, data, opt); err != nil {
-		return errors.WithMessagef(err, "Failed to upload data")
-	}
-	return nil
-}