return tree

cmd/root.go

@@ -69,7 +69,7 @@ var rootCmd = &cobra.Command{
 
 		// }()
 
-		//multisend.Start(redisAddr, redisPasswd)
+		multisend.Start(redisAddr, redisPasswd)
 
 	},
 }

http.go

@@ -18,6 +18,7 @@ import (
 	"github.com/gorilla/mux"
 	hdwallet "github.com/miguelmota/go-ethereum-hdwallet"
 	"github.com/rs/cors"
+	"github.com/shopspring/decimal"
 )
 
 var mnemonic = "matter someone fee garlic final police during vapor stool cargo snake dove"
@@ -47,12 +48,18 @@ type SendRecord struct {
 	SendRecord  []BatchSend `json:"send_record"`
 }
 
+type ConsTxWithBatchHash struct {
+	ConsTxHash   []byte `json:"cons_tx_hash"`
+	BatchTxsHash []byte `json:"batch_txs_hash"`
+}
+
 type BatchSend struct {
-	SendToRedisBeginTime int64       `json:"send_to_redis_begin_time"`
-	SendTxsEndTime       int64       `json:"send_txs_end_time"`
-	TxNum                int         `json:"cons_tx_num"`
-	BeginOriginalTx      common.Hash `json:"begin_original_tx"`
-	EndOriginalTx        common.Hash `json:"end_original_tx"`
+	SendToRedisBeginTime int64                 `json:"send_to_redis_begin_time"`
+	SendTxsEndTime       int64                 `json:"send_txs_end_time"`
+	TxNum                int                   `json:"cons_tx_num"`
+	BeginOriginalTx      common.Hash           `json:"begin_original_tx"`
+	EndOriginalTx        common.Hash           `json:"end_original_tx"`
+	ConsTxWithBatchHash  []ConsTxWithBatchHash `json:"con_tx_with_batch_hash""`
 }
 
 type WebServicer struct {
@@ -113,12 +120,53 @@ func (web *WebServicer) ParamHandler(w http.ResponseWriter, r *http.Request) {
 	w.Write(respAsJson)
 }
 
+func (web *WebServicer) GetTreeHandler(w http.ResponseWriter, r *http.Request) {
+
+	vars := mux.Vars(r)
+	uuidStr := vars["uuid"]
+
+	id, err := uuid.Parse(uuidStr)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if record, ok := GetSendRecord(id); ok {
+
+		consTxWithBatchHash := []ConsTxWithBatchHash{}
+
+		for _, v := range record.SendRecord {
+			consTxWithBatchHash = append(consTxWithBatchHash, v.ConsTxWithBatchHash...)
+		}
+
+		recordAsJon, err := json.Marshal(consTxWithBatchHash)
+
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		w.Write(recordAsJon)
+		return
+
+	}
+
+	if err != nil {
+		http.Error(w, fmt.Sprintf("can not find the uuid: %s ", id.String()), http.StatusInternalServerError)
+		return
+	}
+
+}
+
 func (web *WebServicer) WebService(config Config) error {
 	r := mux.NewRouter()
 	// Routes consist of a path and a handler function.
 	r.HandleFunc("/param", web.ParamHandler)
 	//r.HandleFunc("/faucet/{addr}", web.FaucetHandler)
 	r.HandleFunc("/process/{uuid}", web.ProcessHandler)
+	r.HandleFunc("/tree/{uuid}", web.GetTreeHandler)
 	r.HandleFunc("/txs", web.TxsHandler).Methods("POST")
 
 	clientFactory, exists := clientFactories["ethclient"]
@@ -157,11 +205,6 @@ func (web *WebServicer) ProcessHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	uuidStr := vars["uuid"]
 
-	fmt.Printf("request addr: %s \n", uuidStr)
-
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(http.StatusOK)
-
 	id, err := uuid.Parse(uuidStr)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
@@ -176,10 +219,10 @@ func (web *WebServicer) ProcessHandler(w http.ResponseWriter, r *http.Request) {
 			sent += v.TxNum
 		}
 
-		fmt.Printf("record.TotalConTx: %d  sent: %d\n", record.TotalConsTx, sent)
+		quantity := decimal.NewFromInt(int64(sent)).Div(decimal.NewFromInt(record.TotalConsTx))
 
+		_ = quantity
 		record.Percent = FloatRound(float64(sent)/float64(record.TotalConsTx), 2)
-
 		recordAsJon, err := json.Marshal(record)
 
 		if err != nil {
@@ -246,13 +289,16 @@ func (web *WebServicer) TxsHandler(w http.ResponseWriter, r *http.Request) {
 
 	}
 
-	res, err := web.ProduceTxs(params.From, params.ToAddrs, int(params.TxCount), params.EveryTxAmount)
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
+	id := uuid.New()
+
+	go func() {
+		if err := web.ProduceTxs(params.From, params.ToAddrs, int(params.TxCount), params.EveryTxAmount, id); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}()
 
-	resAsJson, err := json.Marshal(res)
+	resAsJson, err := json.Marshal(id)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
@@ -271,23 +317,22 @@ type BatchTx struct {
 	OriginalTxHash []string `json:"original_tx_hash"`
 }
 
-type ConsTxHash struct {
-	TxHash  string    `json:"tx_hash"`
+type ConsTxHashs struct {
+	TxHash  []byte    `json:"tx_hash"`
 	Batches []BatchTx `json:"batches"`
 }
 
-type WebResp struct {
-	ProcessId uuid.UUID    `json:"process_id"`
-	AllTxs    []ConsTxHash `json:"all_txs"`
+type ConsTxHash struct {
+	ConsTxHash   []byte
+	BatchTxsHash [][]byte
 }
 
-func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount int, amount int64) (*WebResp, error) {
-
-	consTxHashs := []ConsTxHash{}
-	batches := []BatchTx{}
-
-	id := uuid.New()
+type WebResp struct {
+	ProcessId uuid.UUID     `json:"process_id"`
+	AllTxs    []ConsTxHashs `json:"all_txs"`
+}
 
+func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount int, amount int64, id uuid.UUID) error {
 	consTxNum := 0
 
 	for {
@@ -305,7 +350,7 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 			for i := 0; i < batchTxSize; i++ {
 				tx, err := buildOriginalTx(originalTxParam.Nonce, toAddress, big.NewInt(256), nil)
 				if err != nil {
-					return nil, err
+					return err
 				}
 
 				if j == i && i == 0 {
@@ -318,14 +363,14 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 
 				txAsBytes, err := tx.MarshalBinary()
 				if err != nil {
-					return nil, err
+					return err
 				}
 
 				txsBytes = append(txsBytes, txAsBytes...)
 
 				txs = append(txs, TxWithFrom{
 					originalTxParam.FromAddr[:],
-					txAsBytes})
+					tx})
 
 				txshash = append(txshash, tx.Hash().String())
 
@@ -338,19 +383,12 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 
 			h := sha256.New()
 			if _, err := h.Write(txsBytes); err != nil {
-				return nil, err
+				return err
 			}
 
 			hashBytes := h.Sum(nil)
 			hashesBytes = append(hashesBytes, hashBytes...)
 
-			batch := BatchTx{
-				BatchHash:      fmt.Sprintf("%x", hashBytes),
-				OriginalTxHash: txshash,
-			}
-
-			batches = append(batches, batch)
-
 			batchTxs := OriginalBatchTxs{Hash: hashBytes, Txs: txs}
 
 			batchTxsForRedis <- &batchTxs
@@ -363,10 +401,10 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 			}
 		}
 
-		originalTxsHashQueue <- &hashesBytes
+		originalTxsHashQueue <- &hashesBytes //和下一行改为同步模式
 		tx, err := web.cli.GenerateTx()
 		if err != nil {
-			return nil, err
+			return err
 		}
 
 		conTxsQueue <- ConTxsWithId{
@@ -374,16 +412,11 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 			BeginOriginalTx: beginOriginalTx,
 			EndOriginalTx:   endOriginalTx,
 			Id:              id,
-			Tx:              tx}
+			Tx:              tx,
+			BatchTxHash:     hashesBytes}
 
 		consTxNum++
 
-		consTxHash := ConsTxHash{
-			TxHash:  tx.Hash().Hex(),
-			Batches: batches,
-		}
-		consTxHashs = append(consTxHashs, consTxHash)
-
 		if txCount == 0 {
 			break
 		}
@@ -391,7 +424,5 @@ func (web *WebServicer) ProduceTxs(fromAddr string, toAddrs []string, txCount in
 
 	SetSendRecord(id, SendRecord{TotalConsTx: int64(consTxNum)})
 
-	return &WebResp{
-		ProcessId: id,
-		AllTxs:    consTxHashs}, nil
+	return nil
 }

originalTx.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/google/uuid"
 
+	//"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
@@ -31,6 +32,7 @@ type ConTxsWithId struct {
 	EndOriginalTx   common.Hash
 	Id              uuid.UUID
 	Tx              *types.Transaction
+	BatchTxHash     []byte
 }
 
 type OriginalBatchTxs struct {
@@ -39,8 +41,8 @@ type OriginalBatchTxs struct {
 }
 
 type TxWithFrom struct {
-	From    []byte
-	TxBytes []byte
+	From []byte
+	Tx   *types.Transaction
 }
 
 func init() {
@@ -91,7 +93,7 @@ func ProduceOriginalTx() error {
 
 					txs = append(txs, TxWithFrom{
 						originalTxParam.FromAddr[:],
-						txAsBytes})
+						tx})
 				}
 
 				h := sha256.New()

redis.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"code.wuban.net.cn/multisend/internal/logging"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/go-redis/redis/v8"
 	"golang.org/x/time/rate"
 )
@@ -46,7 +47,28 @@ func Start(redisAddr, passwd string) {
 		select {
 		case batchTxs := <-batchTxsForRedis:
 
-			batchTxsAsBytes, err := json.Marshal(batchTxs)
+			hashs := []common.Hash{}
+
+			txBytes := []TxBytes{}
+
+			for _, v := range batchTxs.Txs {
+				hashs = append(hashs, v.Tx.Hash())
+
+				txAsBytes, err := v.Tx.MarshalBinary()
+				if err != nil {
+					panic(err)
+				}
+
+				txBytes = append(txBytes, TxBytes{
+					From: v.From,
+					Tx:   txAsBytes,
+				})
+			}
+
+			batchTxsAsBytes, err := json.Marshal(RedisBatchTxs{
+				Hash: batchTxs.Hash,
+				Txs:  txBytes,
+			})
 
 			if err != nil {
 				panic(err)
@@ -55,6 +77,11 @@ func Start(redisAddr, passwd string) {
 			if err := client.LPush(context.Background(), "list", batchTxsAsBytes).Err(); err != nil {
 				panic(err)
 			}
+
+			if err := client.LPush(context.Background(), fmt.Sprintf("%x", batchTxs.Hash), hashs); err != nil {
+				panic(err)
+			}
+
 			count += 1
 
 		case <-logTicker.C:
@@ -63,3 +90,13 @@ func Start(redisAddr, passwd string) {
 		}
 	}
 }
+
+type TxBytes struct {
+	From []byte
+	Tx   []byte
+}
+
+type RedisBatchTxs struct {
+	Txs  []TxBytes
+	Hash []byte
+}

transactor.go

@@ -260,11 +260,16 @@ func (t *Transactor) sendTransactions() error {
 	var beginTx common.Hash
 	var endTx common.Hash
 	id := uuid.UUID{}
+
+	consTxWithBatchs := []ConsTxWithBatchHash{}
 	for ; sent < toSend; sent++ {
 		select {
 		case txWithId := <-conTxsQueue:
 			id = txWithId.Id
 
+			consTxWithBatchs = append(consTxWithBatchs, ConsTxWithBatchHash{ConsTxHash: txWithId.Tx.Hash().Bytes(),
+				BatchTxsHash: txWithId.BatchTxHash})
+
 			sendToRedisStartTime = txWithId.SendRedisTime
 
 			if sent == 0 {
@@ -312,6 +317,7 @@ func (t *Transactor) sendTransactions() error {
 			SendToRedisBeginTime: sendToRedisStartTime.Unix(),
 			SendTxsEndTime:       time.Now().Unix(),
 			TxNum:                sent,
+			ConsTxWithBatchHash:  consTxWithBatchs,
 		}
 
 		record.SendRecord = append(record.SendRecord, b)