tar-stream-demo

在Docker的save, cp等命令，都使用到了文件打包传输的功能。所以，这次分析将介绍这方面的一个demo，demo的功能如下：
从服务端拷贝：

1. 在服务端读取文件，然后把文件写入数据流中；
2. 在客户端读取数据流中的内容，然后还原成文件。

拷贝到服务端：

1. 在客户端把文件写入数据流；
2. 在服务端读取数据流的内容，然后还原成文件。

将模仿Docker基于”archive/tar”包编写。

从服务端拷贝

server端

package main

import (
	"archive/tar"
	"io"
	"os"
	"time"
	"net/http"
)

func compressFile(file_path string, tw *tar.Writer) error {
	//获取文件相关信息
	fileinfo, err := os.Stat(file_path)
        if err != nil {
                return err
        }
        hdr, err := tar.FileInfoHeader(fileinfo, "")
        if err != nil {
                return err
        }

	//写入头文件信息
        err = tw.WriteHeader(hdr)
        if err != nil {
                return err
        }

        f, err := os.Open(file_path)
        if err != nil {
                return err
        }
	io.Copy(tw, f)
	return nil
}

func compress(path string, tw *tar.Writer) error {
	file, err := os.Open(path)
	if err != nil {
		return err
	}
	info, err := file.Stat()
	if err != nil {
		return err
	}
	if info.IsDir() {
		fileInfos, err := file.Readdir(-1)
		if err != nil {
			return err
		}
		for _, fi := range fileInfos {
			file_path := file.Name() + "/" + fi.Name()
			if err != nil {
				return err
			}
			compress(file_path, tw)
		}
	} else {
		compressFile(path, tw)
	}
	return nil
}

func CompressData(w http.ResponseWriter, req *http.Request) {
	tw := tar.NewWriter(w)
	defer tw.Close()

	compress("/home/fankang/go/demo/tar/data/nginx.tar", tw)
	time.Sleep(5*time.Second)
	compress("/home/fankang/go/demo/tar/data/nginx1.tar", tw)
	time.Sleep(5*time.Second)
	compress("/home/fankang/go/demo/tar/data/nginx2.tar", tw)
	time.Sleep(5*time.Second)
	compress("/home/fankang/go/demo/tar/data/data", tw)
}

func main() {
        http.HandleFunc("/data", CompressData)
        http.ListenAndServe(":8001", nil)
}

其中compress()可以打包文件或文件夹。

client端

package main

import (
	"archive/tar"
	"io"
	"fmt"
	"os"
	"net/http"
)

func main() {
	client := &http.Client{}
        url := "http://127.0.0.1:8001/data"
        request, err := http.NewRequest("GET", url, nil)
        if err != nil {
                panic(err)
        }
        request.Header.Set("Connection", "close")
        response, _ := client.Do(request)
	r := tar.NewReader(response.Body)
	for hdr, err := r.Next(); err != io.EOF; hdr, err = r.Next() {
		if err != nil {
			fmt.Println(err)
			return
		}
		fileinfo := hdr.FileInfo()
		fmt.Println(fileinfo.Name())
		f, err := os.Create("/home/fankang/go/demo/tar/newdata/" + fileinfo.Name())
		if err != nil {
			fmt.Println(err)
		}
		defer f.Close()
		_, err = io.Copy(f, r)
		if err != nil {
			fmt.Println(err)
		}
	}
	response.Body.Close()
}

client端把response.Body封装成tar.ReaderCloser，然后调用Next()获取打包内容。在demo中，client接收到的所有内容将被放在”/home/fankang/go/demo/tar/newdata/“目录下。

拷贝到服务端

server端

package main

import (
	"archive/tar"
	"io"
	"fmt"
	"os"
	"net/http"
)

func DeCompressData(w http.ResponseWriter, req *http.Request) {
	r := tar.NewReader(req.Body)
        for hdr, err := r.Next(); err != io.EOF; hdr, err = r.Next() {
                if err != nil {
                        fmt.Println(err)
                        return
                }
                fileinfo := hdr.FileInfo()
                fmt.Println(fileinfo.Name())
                f, err := os.Create("/home/fankang/go/demo/tar/newdata/" + fileinfo.Name())
                if err != nil {
                        fmt.Println(err)
                }
                defer f.Close()
                _, err = io.Copy(f, r)
                if err != nil {
                        fmt.Println(err)
                }
        }
}

func main() {
	http.HandleFunc("/data", DeCompressData)
        http.ListenAndServe(":8001", nil)
}

client端

package main

import (
	"archive/tar"
	"io"
	"os"
	"fmt"
	"net/http"
)

func compressFile(file_path string, tw *tar.Writer) error {
	//获取文件相关信息
	fileinfo, err := os.Stat(file_path)
        if err != nil {
                return err
        }
        hdr, err := tar.FileInfoHeader(fileinfo, "")
        if err != nil {
                return err
        }

	//写入头文件信息
        err = tw.WriteHeader(hdr)
        if err != nil {
                return err
        }

        f, err := os.Open(file_path)
        if err != nil {
                return err
        }
	io.Copy(tw, f)
	return nil
}

func compress(path string, tw *tar.Writer) error {
	file, err := os.Open(path)
	if err != nil {
		return err
	}
	info, err := file.Stat()
	if err != nil {
		return err
	}
	if info.IsDir() {
		fileInfos, err := file.Readdir(-1)
		if err != nil {
			return err
		}
		for _, fi := range fileInfos {
			file_path := file.Name() + "/" + fi.Name()
			if err != nil {
				return err
			}
			compress(file_path, tw)
		}
	} else {
		compressFile(path, tw)
	}
	return nil
}

func main() {
	client := &http.Client{}
        url := "http://127.0.0.1:8001/data"
	pipeReader, pipeWriter := io.Pipe()
	tw := tar.NewWriter(pipeWriter)
	go func() {
		compress("/home/fankang/go/demo/tar/data/data", tw)
		defer tw.Close()
	}()
	request, err := http.NewRequest("PUT", url, pipeReader)
        if err != nil {
                panic(err)
        }
	request.Header.Set("Connection", "close")
        response, _ := client.Do(request)
	response.Body.Close()
}