[rget] add parallelDownload

Author: imura81gt

kadai3/imura81gt/rget/.go-version

@@ -0,0 +1 @@
+1.13.4

kadai3/imura81gt/rget/README.md

@@ -7,3 +7,42 @@ Command
 ```
 go run cmd/rget/main.go https://upload.wikimedia.org/wikipedia/commons/1/16/Notocactus_minimus.jpg
 ```
+
+Theme
+-----------------------------------------
+
+分割ダウンロードを行う
+
+元ネタ: https://qiita.com/codehex/items/d0a500ac387d39a34401
+
+- [x]Rangeアクセスを用いる
+- [ ]いくつかのゴルーチンでダウンロードしてマージする
+- [x]エラー処理を工夫する
+- [x]golang.org/x/sync/errgroupパッケージなどを使ってみる
+- [x]キャンセルが発生した場合の実装を行う
+
+ref: https://qiita.com/codehex/items/d0a500ac387d39a34401
+
+
+
+Note.
+------------------------------------------
+
+### Range Request
+
+https://developer.mozilla.org/ja/docs/Web/HTTP/Range_requests
+
+> Accept-Ranges が HTTP レスポンスに存在した場合 (そして値が "none" ではない場合)、サーバーは範囲リクエストに対応しています。これは例えば、 HEAD リクエストを cURL で発行することで確認することができます。
+
+
+https://developer.mozilla.org/ja/docs/Web/HTTP/Headers/Accept-Ranges
+
+> Accept-Ranges: bytes
+> Accept-Ranges: none
+
+https://developer.mozilla.org/ja/docs/Web/HTTP/Headers/Range
+
+> Range: <unit>=<range-start>-
+> Range: <unit>=<range-start>-<range-end>
+> Range: <unit>=<range-start>-<range-end>, <range-start>-<range-end>
+> Range: <unit>=<range-start>-<range-end>, <range-start>-<range-end>, <range-start>-<range-end>

kadai3/imura81gt/rget/go.mod

@@ -0,0 +1,5 @@
+module github.com/gopherdojo/dojo7/kadai3/imura81gt/rget
+
+go 1.13
+
+require golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e

kadai3/imura81gt/rget/go.sum

@@ -0,0 +1,2 @@
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

kadai3/imura81gt/rget/rget.go

@@ -1,78 +1,183 @@
 package rget
 
 import (
+	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"os"
+	"path"
+
+	"golang.org/x/sync/errgroup"
 )
 
 type Option struct {
-	Concurrency int
-	Url         string
+	Concurrency   int
+	URL           string
+	OutputDir     string
+	ContentLength int64
+	Units         Units
 }
 
 type Unit struct {
-	RangeStart int64
-	RangeEnd   int64
+	RangeStart   int64
+	RangeEnd     int64
+	TempFileName string
 }
 
 type Units []Unit
 
 func Run(option Option) {
 	fmt.Printf("%+v\n", option)
-	cl, err := contentLength(option.Url)
+	err := option.contentLength()
 	if err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(1)
+		fmt.Errorf("%s", err)
 	}
 
-	units := divide(cl, option.Concurrency)
+	option.divide()
 
-	//TODO: check errors
-	download(units)
+	err = option.parallelDownload()
+	if err != nil {
+		fmt.Errorf("%s", err)
+	}
 
 }
 
-func contentLength(url string) (int64, error) {
-	resp, err := http.Head(url)
+func (o *Option) contentLength() error {
+	//resp, err := http.Head(url)
+	resp, err := http.Head(o.URL)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
-		return 0, err
+		//return 0, err
+		return err
 	}
 
 	if resp.Header.Get("Accept-Ranges") == "" {
-		err := fmt.Errorf("This URL cannot support Ranges Requests")
+		err := fmt.Errorf("%s URL cannot support Ranges Requests", o.URL)
 		// fmt.Fprintln(os.Stderr, err)
-		return resp.ContentLength, err
+		//return resp.ContentLength, err
+		return err
 	}
 	if resp.Header["Accept-Ranges"][0] == "none" {
-		err := fmt.Errorf("This URL cannot support Ranges Requests")
+		err := fmt.Errorf("%s cannot support Ranges Requests", o.URL)
 		// fmt.Fprintln(os.Stderr, err)
-		return resp.ContentLength, err
+		//return resp.ContentLength, err
+		return err
 	}
 	if resp.ContentLength == 0 {
-		err := fmt.Errorf("This URL size is %i", resp.Header["Content-Length"][0])
+		err := fmt.Errorf("%s size is %s", o.URL, resp.Header["Content-Length"][0])
 		// fmt.Fprintln(os.Stderr, err)
-		return resp.ContentLength, err
+		//return resp.ContentLength, err
+		return err
 	}
 
-	return resp.ContentLength, nil
+	o.ContentLength = resp.ContentLength
+	//return resp.ContentLength, nil
+	return err
 }
 
-func divide(contentLength int64, concurrency int) Units {
+//func divide(contentLength int64, concurrency int) Units {
+func (o *Option) divide() {
 	var units []Unit
 
-	sbyte := contentLength / int64(concurrency)
-	for i := 0; i < concurrency; i++ {
+	//sbyte := contentLength / int64(concurrency)
+	sbyte := o.ContentLength / int64(o.Concurrency)
+
+	//	for i := 0; i < concurrency; i++ {
+	for i := 0; i < o.Concurrency; i++ {
 		units = append(units, Unit{
-			RangeStart: int64(i) * sbyte,
-			RangeEnd:   int64((i+1))*sbyte - 1,
+			RangeStart:   int64(i) * sbyte,
+			RangeEnd:     int64((i+1))*sbyte - 1,
+			TempFileName: fmt.Sprintf("%d_%s", i, path.Base(o.URL)),
+		})
+	}
+
+	o.Units = units
+	//return units
+}
+
+// func download(units Units) {
+// 	filepath.Split()
+// 	fmt.Println(units)
+// }
+
+func (o *Option) parallelDownload() error {
+	fmt.Println("parallelDownload", o.Units)
+
+	eg, ctx := errgroup.WithContext(context.Background())
+	for i := range o.Units {
+		// https://godoc.org/golang.org/x/sync/errgroup#example-Group--Parallel
+		// https://golang.org/doc/faq#closures_and_goroutines
+		i := i
+		eg.Go(func() error {
+			return o.downloadWithContext(ctx, i)
 		})
 	}
 
-	return units
+	if err := eg.Wait(); err != nil {
+		o.clearCache()
+		return err
+	}
+
+	return nil
+}
+
+func (o *Option) downloadWithContext(ctx context.Context, i int) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	fmt.Printf("Downloading: %v %+v\n", i, o.Units[i])
+
+	//v1.13
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, o.URL, nil)
+	if err != nil {
+		return fmt.Errorf("Error: %v", err)
+	}
+
+	// add range header
+	fmt.Printf(fmt.Sprintf("bytes=%d-%d", o.Units[i].RangeStart, o.Units[i].RangeEnd))
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", o.Units[i].RangeStart, o.Units[i].RangeEnd))
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	defer resp.Body.Close()
+	if err != nil {
+		return fmt.Errorf("Error: %v", err)
+	}
+
+	select {
+	case <-ctx.Done():
+		fmt.Printf("Done: %v %+v\n", i, o.Units[i])
+		return nil
+	default:
+		fmt.Println("default:", i, o.Units[i])
+		//return fmt.Errorf("Error: %v %+v", i, o.Units[i])
+	}
+
+	w, err := os.Create(o.Units[i].TempFileName)
+	if err != nil {
+		return fmt.Errorf("Error: %v", err)
+	}
+	defer func() error {
+		if err := w.Close(); err != nil {
+			return fmt.Errorf("Error: %v", err)
+		}
+		return nil
+	}()
+
+	_, err = io.Copy(w, resp.Body)
+	if err != nil {
+		return fmt.Errorf("Error: %v", err)
+	}
+
+	return nil
+}
+
+func (o *Option) conbine() error {
+	return nil
 }
 
-func download(units Units) {
-	fmt.Println(units)
+func (o *Option) clearCache() error {
+	//TODO: remove temporary files
+	return nil
 }

kadai3/imura81gt/rget/rget_test.go

@@ -0,0 +1 @@
+package rget