copy rewrite-2 to master

Author: HusseinElguindi

.gitignore

@@ -0,0 +1,2 @@
+
+*.mp4

README.md

@@ -12,10 +12,10 @@ go get github.com/husseinelguindi/mtd
 - [x] HTTP/HTTPS download support
 - [x] Offset write to file
 - [x] Synchronized writing to file (never write 2 things at once)
-- [x] Go module interface to download within your program
+- [x] Go module/library interface
+- [x] Multiple downloads in a single instance
 - [ ] Pause/resume support
-- [ ] Multiple downloads in a single instance
-- [ ] Command line interface (coming soon)
+- [ ] Command line interface
     - [ ] Args
     - [ ] Console input
 
@@ -24,3 +24,4 @@ go get github.com/husseinelguindi/mtd
 A writer Goroutine that only writes one chunk at a time, this was done for a number of a simple reason:
 - Seeking to a location in a file, physically moves the write head of a hard drive, which slows the writing to a file.
 - This means that writing chunks, that are near each other, would yield better performance, avoiding the hard drive seek delay.
+- This also means that using a bigger chunk/buffer size is beneficial to write speeds (less system calls, as well).

go.mod

@@ -0,0 +1,3 @@
+module github.com/husseinelguindi/mtd
+
+go 1.16

http.go

@@ -0,0 +1,138 @@
+package mtd
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+var (
+	ErrNoContentLength = errors.New("content length is unset or zero")
+)
+
+type metadata struct {
+	contentLength    int64
+	contentType      string
+	acceptByteRanges bool
+}
+
+// TODO: implement download retries
+// TODO: progress bars
+// TODO: background program that connects to terminal command (like Docker cli)
+
+func (t Task) httpHEAD() (metadata, error) {
+	// Create request
+	req, _ := http.NewRequest(http.MethodHead, t.URL, nil)
+	// Set task's request headers
+	for k, v := range t.Headers {
+		req.Header.Set(k, v)
+	}
+
+	// Perform the HEAD request with the task's client
+	resp, err := t.Client.Do(req)
+	if err != nil {
+		return metadata{}, err
+	}
+	defer resp.Body.Close()
+
+	// Parse headers and create metadata
+	return metadata{
+		contentLength:    resp.ContentLength,
+		contentType:      resp.Header.Get("Content-Type"),
+		acceptByteRanges: strings.EqualFold(resp.Header.Get("Accept-Ranges"), "Bytes"),
+	}, err
+}
+
+func (t Task) Download() error {
+	// Get metadata from a HEAD request
+	meta, err := t.httpHEAD()
+	if err != nil {
+		return err
+	}
+	// Ensure the content has a positive, non-zero length
+	if meta.contentLength <= 0 {
+		return ErrNoContentLength
+	}
+
+	// Determine the chunk size taking into consideration the content length, number of chunks,
+	// and whether or not the file server allows for byte ranges
+	chunkSize := meta.contentLength
+	if meta.acceptByteRanges && t.Chunks > 1 {
+		chunkSize = meta.contentLength / int64(t.Chunks) // Floored
+	}
+
+	// Prepare waitgroup with the expected number of goroutines
+	wg := &sync.WaitGroup{}
+	wg.Add(int(t.Chunks))
+
+	// Initialize the error channel
+	errc := make(chan error)
+
+	// Distibute the byte ranges between t.Chunks number of goroutines
+	var start, end int64
+	for i := uint(1); i < t.Chunks; i++ {
+		end += chunkSize
+		go t.httpWorker(byteRange{start, end - 1}, wg, errc)
+		start = end
+	}
+	// Handle remaining bytes (or all bytes if t.Chunks is 1)
+	end += meta.contentLength - start
+	go t.httpWorker(byteRange{start, end}, wg, errc)
+
+	// Listen for errors
+	go func() {
+		for {
+			err := <-errc
+			// TODO: handle errors
+			fmt.Println("error!", err)
+		}
+	}()
+
+	// Wait for worker goroutines to finish
+	wg.Wait()
+	return nil
+}
+
+func (t Task) httpWorker(bRange byteRange, wg *sync.WaitGroup, errc chan<- error) {
+	defer wg.Done()
+
+	// Get body
+	rc, err := t.httpGET(bRange)
+	if err != nil {
+		errc <- err
+		return
+	}
+	defer rc.Close()
+
+	// Write the body
+	_, err = t.write(rc, bRange)
+	if err != nil {
+		errc <- err
+		return
+	}
+}
+
+type byteRange struct{ start, end int64 }
+
+func (b byteRange) Header() string { return fmt.Sprintf("bytes=%d-%d", b.start, b.end) }
+func (b byteRange) Valid() bool    { return b.end > 0 && b.end > b.start }
+
+func (t Task) httpGET(bRange byteRange) (io.ReadCloser, error) {
+	req, _ := http.NewRequest(http.MethodGet, t.URL, nil)
+	// Set task's request headers
+	for k, v := range t.Headers {
+		req.Header.Set(k, v)
+	}
+
+	// Set the byte range header if a range is passed
+	if bRange.Valid() {
+		req.Header.Set("Range", bRange.Header())
+	}
+
+	// Perform the GET request with the task's client
+	resp, err := t.Client.Do(req)
+	return resp.Body, err
+}

http_test.go

@@ -0,0 +1,36 @@
+package mtd
+
+import (
+	"context"
+	"os"
+	"runtime"
+	"testing"
+)
+
+func TestHTTPDownload(t *testing.T) {
+	f, err := os.OpenFile("./vid.mp4", os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(0666))
+	if err != nil {
+		t.FailNow()
+	}
+	defer f.Close()
+
+	writer := Writer{}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	task := Task{
+		URL:     "https://file-examples-com.github.io/uploads/2017/04/file_example_MP4_1920_18MG.mp4",
+		Chunks:  uint(runtime.NumCPU()),
+		BufSize: 7 * 1024 * 1024, // ~7mb
+
+		Dst:    f,
+		Writer: &writer,
+
+		Ctx: ctx,
+	}
+
+	if err := task.Download(); err != nil {
+		t.FailNow()
+	}
+}

task.go

@@ -0,0 +1,79 @@
+package mtd
+
+import (
+	"context"
+	"io"
+	"net/http"
+)
+
+type Task struct {
+	URL string
+	// Threads uint
+	// ChunkSize int64
+	Chunks  uint
+	BufSize int64
+
+	Client  http.Client
+	Headers map[string]string
+
+	Dst    io.WriterAt
+	Writer *Writer
+
+	Ctx context.Context
+}
+
+func (t Task) write(rc io.ReadCloser, bRange byteRange) (written int, err error) {
+	// Determine the smallest useable byte buffer size and allocate the space for it
+	bufSize := minInt64(t.BufSize, bRange.end-bRange.start+1)
+	buf := make([]byte, bufSize)
+
+	// Prepare backet for reusing
+	packet := packet{
+		buf: buf,
+		dst: t.Dst,
+	}
+
+	// Set offset for chunked downloads
+	if bRange.Valid() {
+		packet.off = bRange.start
+	}
+
+	for {
+		// Handle cancel without blocking
+		select {
+		case <-t.Ctx.Done():
+			return
+		default:
+		}
+
+		// Read up to bufSize bytes
+		var n int
+		n, err = io.ReadFull(rc, buf[:])
+		if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
+			return
+		}
+
+		// Break if no bytes were read (no bytes left to write)
+		if n == 0 {
+			break
+		}
+
+		// Set the packet's buffer as a slice of the newly read bytes
+		packet.buf = buf[:n]
+
+		// Write the packet
+		n, err = t.Writer.Write(packet)
+
+		// Update offset for next write
+		packet.off += int64(n)
+		// Update the total number of bytes written
+		written += n
+
+		// Handle any write errors
+		if err != nil {
+			return
+		}
+	}
+
+	return written, nil
+}

utils.go

@@ -1,13 +1,8 @@
-package mtd
-
-import "net/http"
-
-func setReqHeaders(req *http.Request, headers http.Header) {
-	if headers == nil {
-		return
-	}
-
-	for k, v := range headers {
-		req.Header.Set(k, v[0])
-	}
-}
+package mtd
+
+func minInt64(x, y int64) int64 {
+	if x < y {
+		return x
+	}
+	return y
+}

writer.go

@@ -2,64 +2,24 @@ package mtd
 
 import (
 	"io"
+	"sync"
 )
 
-type writeObj struct {
-	dst    io.WriterAt
-	buf    []byte
-	offset uint64
-	errCh  chan error
-}
-
 type Writer struct {
-	writeCh chan writeObj
-	closeCh chan int
-}
-
-// NewWriter initializes a new writer
-func NewWriter() Writer {
-	w := Writer{
-		writeCh: make(chan writeObj),
-		closeCh: make(chan int, 1),
-	}
-	w.listen()
-	return w
+	sync.Mutex
 }
 
-func (w Writer) listen() {
-	go func() {
-		for {
-			select {
-			case writeObj := <-w.writeCh:
-				writeObj.write()
-			case code := <-w.closeCh:
-				if code == 0 {
-					w.closeCh <- 0
-				}
-
-				break
-			}
-		}
-	}()
+type packet struct {
+	dst io.WriterAt
+	buf []byte
+	off int64
 }
 
-func (wo writeObj) write() {
-	// if _, err := wo.dest.Seek(int64(wo.offset), io.SeekStart); err != nil {
-	// 	wo.errCh <- err
-	// 	return
-	// }
-
-	wrote := 0
-	for wrote < len(wo.buf) {
-		// n, err := wo.dest.Write(wo.buf)
-		n, err := wo.dst.WriteAt(wo.buf, int64(wo.offset+uint64(wrote)))
-		if err != nil {
-			wo.errCh <- err
-			return
-		}
-		wrote += n
-	}
+func (w *Writer) Write(p packet) (int, error) {
+	// Block until free
+	w.Lock()
+	defer w.Unlock()
 
-	wo.errCh <- nil
-	// close(wo.errCh)
+	// Write buf at offset
+	return p.dst.WriteAt(p.buf[:], p.off)
 }