add BroadcastData

Author: outprog

client.go

@@ -54,15 +54,28 @@ func (c *Client) GetInfo() (info *types.NetworkInfo, err error) {
 	return
 }
 
-func (c *Client) Peers() ([]string, error) {
+func (c *Client) GetPeers() ([]string, error) {
 	body, _, err := c.httpGet("peers")
 	if err != nil {
 		return nil, ErrBadGateway
 	}
 
 	peers := make([]string, 0)
 	err = json.Unmarshal(body, &peers)
-	return peers, err
+	if err != nil {
+		return nil, err
+	}
+
+	// filter local
+	fpeers := make([]string, 0)
+	for _, p := range peers {
+		if strings.Contains(p, "127.0.0.") {
+			continue
+		}
+		fpeers = append(fpeers, p)
+	}
+
+	return peers[:], nil
 }
 
 // GetTransactionByID status: Pending/Invalid hash/overspend
@@ -435,42 +448,40 @@ func (c *Client) DownloadChunkData(id string) ([]byte, error) {
 }
 
 func (c *Client) GetTxDataFromPeers(txId string) ([]byte, error) {
-	peers, err := c.Peers()
+	peers, err := c.GetPeers()
 	if err != nil {
 		return nil, err
 	}
+
 	for _, peer := range peers {
-		if strings.Contains(peer, "127.0") {
-			continue
-		}
-		arNode := NewClient("http://" + peer)
-		data, err := arNode.GetTransactionData(txId)
+		pNode := NewClient("http://" + peer)
+		data, err := pNode.GetTransactionData(txId)
 		if err != nil {
-			fmt.Printf("get tx data error:%v, peer: %s\n", err, peer)
+			log.Error("get tx data failed", "error", err, "peer", peer)
 			continue
 		}
 		return data, nil
 	}
+
 	return nil, errors.New("get tx data from peers failed")
 }
 
-func (c *Client) UploadTxDataToPeers(txId string, data []byte) error {
-	peers, err := c.Peers()
+func (c *Client) BroadcastData(txId string, data []byte, numOfNodes int64) error {
+	peers, err := c.GetPeers()
 	if err != nil {
 		return err
 	}
 
-	count := 0
+	count := int64(0)
 	for _, peer := range peers {
-		if strings.Contains(peer, "127.0") {
-			continue
-		}
+
 		fmt.Printf("upload peer: %s, count: %d\n", peer, count)
 		arNode := NewClient("http://" + peer)
 		uploader, err := CreateUploader(arNode, txId, data)
 		if err != nil {
 			continue
 		}
+
 	Loop:
 		for !uploader.IsComplete() {
 			if err := uploader.UploadChunk(); err != nil {
@@ -483,10 +494,11 @@ func (c *Client) UploadTxDataToPeers(txId string, data []byte) error {
 		if uploader.IsComplete() { // upload success
 			count++
 		}
-		if count > 20 {
+		if count >= numOfNodes {
 			return nil
 		}
 	}
+
 	return fmt.Errorf("upload tx data to peers failed, txId: %s", txId)
 }
 

client_test.go

@@ -196,30 +196,30 @@ func TestClient_GetTransactionDataByGateway(t *testing.T) {
 func TestClient_GetPeers(t *testing.T) {
 	arNode := "https://arweave.net"
 	cli := NewClient(arNode)
-	peers, err := cli.Peers()
+	peers, err := cli.GetPeers()
 	assert.NoError(t, err)
 	t.Log(len(peers))
 }
 
-func Test_UploadTxDataToPeers(t *testing.T) {
-	data := make([]byte, 1103732)
-	data[0] = byte('z')
-	data[1] = byte('y')
-	data[2] = byte('j')
-	for i := 3; i < len(data); i++ {
-		data[i] = byte('a')
-	}
-
-	cli := NewClient("https://arweave.net")
-	txId := "D3GOny9cItUEc8qAl1oLUtnoLOB3OfSB-wKbw8TUIRc"
-	err = cli.UploadTxDataToPeers(txId, data)
-	assert.NoError(t, err)
-}
+// func Test_BroadcastData(t *testing.T) {
+// 	data := make([]byte, 1103732)
+// 	data[0] = byte('z')
+// 	data[1] = byte('y')
+// 	data[2] = byte('j')
+// 	for i := 3; i < len(data); i++ {
+// 		data[i] = byte('a')
+// 	}
+
+// 	cli := NewClient("https://arweave.net")
+// 	txId := "D3GOny9cItUEc8qAl1oLUtnoLOB3OfSB-wKbw8TUIRc"
+// 	err = cli.BroadcastData(txId, data, 1)
+// 	assert.NoError(t, err)
+// }
 
-func Test_GetTxDataFromPeers(t *testing.T) {
-	cli := NewClient("https://arweave.net")
-	txId := "D3GOny9cItUEc8qAl1oLUtnoLOB3OfSB-wKbw8TUIRc"
-	data, err := cli.GetTxDataFromPeers(txId)
-	assert.NoError(t, err)
-	assert.Equal(t, 1471643, len(data))
-}
+// func Test_GetTxDataFromPeers(t *testing.T) {
+// 	cli := NewClient("https://arweave.net")
+// 	txId := "D3GOny9cItUEc8qAl1oLUtnoLOB3OfSB-wKbw8TUIRc"
+// 	data, err := cli.GetTxDataFromPeers(txId)
+// 	assert.NoError(t, err)
+// 	assert.Equal(t, 1471643, len(data))
+// }