Golang compress-gzip.Reader类(方法)实例源码

func NewDscin(waiter *sync.WaitGroup, filename string) *Dscin {
	d := new(Dscin)
	d.Operation.Waiter = waiter
	d.Operation.Waiter.Add(1)
	file, err := os.Open(filename)
	if err != nil {
		return nil
	}
	d.closer = func() {
		file.Close()
	}
	var reader io.ReadCloser = file
	var uncompressor *gzip.Reader
	if strings.HasSuffix(filename, ".gz") {
		uncompressor, err = gzip.NewReader(file)
		d.closer = func() { uncompressor.Close(); file.Close() }
		reader = uncompressor
	}
	uncompressed_name := strings.TrimRight(filename, ".gz")
	switch {
	case strings.HasSuffix(uncompressed_name, ".gob"):
		d.marshaler = new(formats.GobMarshaler)
	case strings.HasSuffix(uncompressed_name, ".xml"):
		d.marshaler = new(formats.XmlMarshaler)
	}
	if d.marshaler != nil {
		d.marshaler.ValidateFile(reader)
	}
	return d
}

func handleAssets(w http.ResponseWriter, r *http.Request) {
	assets := auto.Assets()
	path := r.URL.Path[1:]
	if path == "" {
		path = "index.html"
	}

	bs, ok := assets[path]
	if !ok {
		w.WriteHeader(http.StatusNotFound)
		return
	}

	mtype := mimeTypeForFile(path)
	if len(mtype) != 0 {
		w.Header().Set("Content-Type", mtype)
	}

	if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
		w.Header().Set("Content-Encoding", "gzip")
	} else {
		// ungzip if browser not send gzip accepted header
		var gr *gzip.Reader
		gr, _ = gzip.NewReader(bytes.NewReader(bs))
		bs, _ = ioutil.ReadAll(gr)
		gr.Close()
	}
	w.Header().Set("Content-Length", fmt.Sprintf("%d", len(bs)))

	w.Write(bs)
}

// GetPage is an HTTP client that automatically decodes gzip when necessary.
func GetPage(url string) ([]byte, error) {
	tr := &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
	}
	client := &http.Client{Transport: tr}
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Add("Accept-Encoding", "gzip, deflate")
	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	var body []byte
	if resp.Header.Get("Content-Encoding") == "gzip" {
		var gz *gzip.Reader
		gz, err = gzip.NewReader(resp.Body)
		if err != nil {
			return nil, err
		}
		defer gz.Close()
		body, err = ioutil.ReadAll(gz)
	} else {
		body, err = ioutil.ReadAll(resp.Body)
	}
	if err != nil {
		return nil, err
	}
	return body, nil
}

func ringOrBuilder(fileName string) (r ring.Ring, b *ring.Builder, err error) {
	var f *os.File
	if f, err = os.Open(fileName); err != nil {
		return
	}
	var gf *gzip.Reader
	if gf, err = gzip.NewReader(f); err != nil {
		return
	}
	header := make([]byte, 16)
	if _, err = io.ReadFull(gf, header); err != nil {
		return
	}
	if string(header[:5]) == "RINGv" {
		gf.Close()
		if _, err = f.Seek(0, 0); err != nil {
			return
		}
		r, err = ring.LoadRing(f)
	} else if string(header[:12]) == "RINGBUILDERv" {
		gf.Close()
		if _, err = f.Seek(0, 0); err != nil {
			return
		}
		b, err = ring.LoadBuilder(f)
	}
	return
}

// readBytes reads an encrypted/compressed steam from an io.Reader
// and returns a decoded byte slice
func readBytes(in io.Reader, key string) ([]byte, error) {
	var gzReader *gzip.Reader  // compressed reader
	var iv [aes.BlockSize]byte // initialization vector
	var cb cipher.Block        // cipher block interface
	var outBytes *bytes.Buffer // output buffer
	var err error              // general error handler

	// init cipher block
	if cb, err = aes.NewCipher(hashKey(key)); err != nil {
		return nil, err
	}

	// init encrypted reader
	encReader := &cipher.StreamReader{
		S: cipher.NewOFB(cb, iv[:]),
		R: in,
	}

	// wrap encrypted reader
	if gzReader, err = gzip.NewReader(encReader); err != nil {
		return nil, err
	}

	// close when done
	defer gzReader.Close()

	// init output
	outBytes = new(bytes.Buffer)

	// read data into output buffer decompressing and decrypting along the way
	_, err = io.Copy(outBytes, gzReader)

	// return bytes and last error state
	return outBytes.Bytes(), err
}

// Decode returns the decoded data for the bundle entry. Returns a
// slice of bytes with the decoded, decompressed (if required), ready
// to use entry data, and an error indication which is not-nil if the
// data cannot be decoded. If argument "flag" is NODC, and the entry
// data are compressed (Entry.Gzip == true), Decode will not
// decompress the data it returns (it will only decode them). In most
// cases it is preferable to use the Reader interface instead of
// calling Decode.
func (e *Entry) Decode(flag int) ([]byte, error) {
	var rs *strings.Reader
	var r64 io.Reader
	var rz *gzip.Reader
	var buf *bytes.Buffer
	var err error

	rs = strings.NewReader(e.Data)
	r64 = base64.NewDecoder(base64.StdEncoding, rs)
	if e.Gzip && (flag&NODC == 0) {
		rz, err = gzip.NewReader(r64)
		if err != nil {
			return nil, err
		}
		defer rz.Close()
	} else {
		rz = nil
	}
	buf = new(bytes.Buffer)
	if rz != nil {
		_, err = io.Copy(buf, rz)
	} else {
		_, err = io.Copy(buf, r64)
	}
	if err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

// Reads transfer-encoding: chunked payloads from the connection reader.
func (c *Connection) readChunkedData() error {
	var err error
	var line []byte
	var size uint64
	var start time.Time

	start = time.Now()
	writer := &nonEmptyWriter{os.Stdout}

	var buffer *bytes.Buffer
	var decompressor *gzip.Reader
	var zipReader *bufio.Reader
	var data []byte

	if c.conf.GZip == true {
		buffer = bytes.NewBufferString("")
	}

	for err == nil {
		line, _, err = c.reader.ReadLine()
		if err != nil {
			return err
		}
		size, err = decodeHexString(line)
		if err != nil {
			str := fmt.Sprintf("Expected hex, got %v", string(line))
			return errors.New(str)
		}
		if c.conf.GZip == false {
			_, err = io.CopyN(writer, c.reader, int64(size))
		} else {
			_, err = io.CopyN(buffer, c.reader, int64(size))
			if err != nil {
				return err
			}
			if decompressor == nil {
				decompressor, err = gzip.NewReader(buffer)
				defer decompressor.Close()
				if err != nil {
					return err
				}
				zipReader = bufio.NewReader(decompressor)
			}
			data = make([]byte, 512, 512)
			_, err = zipReader.Read(data)
			if err != nil {
				return err
			}
			strBuffer := bytes.NewBuffer(data)
			io.CopyN(writer, strBuffer, int64(len(data)))
		}
		if c.conf.TTL > 0 {
			if time.Now().Sub(start).Nanoseconds() > c.conf.TTL {
				return nil
			}
		}
	}
	return err
}

func (s embeddedStatic) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	file := r.URL.Path

	if file[0] == '/' {
		file = file[1:]
	}

	if len(file) == 0 {
		file = "index.html"
	}

	if s.assetDir != "" {
		p := filepath.Join(s.assetDir, filepath.FromSlash(file))
		_, err := os.Stat(p)
		if err == nil {
			http.ServeFile(w, r, p)
			return
		}
	}

	s.mut.RLock()
	theme := s.theme
	modified := s.lastModified
	s.mut.RUnlock()

	bs, ok := s.assets[theme+"/"+file]
	if !ok {
		bs, ok = s.assets[config.DefaultTheme+"/"+file]
		if !ok {
			http.NotFound(w, r)
			return
		}
	}

	if modifiedSince, err := time.Parse(r.Header.Get("If-Modified-Since"), http.TimeFormat); err == nil && modified.Before(modifiedSince) {
		w.WriteHeader(http.StatusNotModified)
		return
	}

	mtype := s.mimeTypeForFile(file)
	if len(mtype) != 0 {
		w.Header().Set("Content-Type", mtype)
	}
	if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
		w.Header().Set("Content-Encoding", "gzip")
	} else {
		// ungzip if browser not send gzip accepted header
		var gr *gzip.Reader
		gr, _ = gzip.NewReader(bytes.NewReader(bs))
		bs, _ = ioutil.ReadAll(gr)
		gr.Close()
	}
	w.Header().Set("Content-Length", fmt.Sprintf("%d", len(bs)))
	w.Header().Set("Last-Modified", modified.Format(http.TimeFormat))
	w.Header().Set("Cache-Control", "public")

	w.Write(bs)
}

// extract 'control' file from 'reader'. the contents of a 'control' file
// is a set of key-value pairs as described in
// https://www.debian.org/doc/debian-policy/ch-controlfields.html
func extractControlFromIpk(reader io.Reader) (string, error) {

	var (
		arReader  *ar.Reader
		tarReader *tar.Reader
		gzReader  *gzip.Reader
	)

	arReader = ar.NewReader(reader)
	for {
		header, err := arReader.Next()
		if err != nil && err != io.EOF {
			return "", fmt.Errorf("extracting contents: %v", err)
		} else if header == nil {
			break
		}

		// NOTE: strangeley the name of the files end with a "/" ... content error?
		if header.Name == "control.tar.gz/" || header.Name == "control.tar.gz" {
			gzReader, err = gzip.NewReader(arReader)
			if err != nil {
				return "", fmt.Errorf("analyzing control.tar.gz: %v", err)
			}
			break
		}
	}

	if gzReader == nil {
		return "", fmt.Errorf("missing control.tar.gz entry")
	}
	defer gzReader.Close()

	buffer := bytes.NewBuffer(nil)
	tarReader = tar.NewReader(gzReader)
	for {
		header, err := tarReader.Next()
		if err != nil && err != io.EOF {
			return "", fmt.Errorf("extracting control.tar.gz: %v", err)
		} else if header == nil {
			break
		}
		if header.Name != "./control" {
			continue
		}

		io.Copy(buffer, tarReader)
		break
	}

	if buffer.Len() == 0 {
		return "", fmt.Errorf("missing or empty 'control' file inside 'control.tar.gz'")
	}
	return buffer.String(), nil
}

func (s embeddedStatic) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	file := r.URL.Path

	if file[0] == '/' {
		file = file[1:]
	}

	if len(file) == 0 {
		file = "index.html"
	}

	if s.assetDir != "" {
		p := filepath.Join(s.assetDir, filepath.FromSlash(file))
		_, err := os.Stat(p)
		if err == nil {
			http.ServeFile(w, r, p)
			return
		}
	}

	bs, ok := s.assets[file]
	if !ok {
		http.NotFound(w, r)
		return
	}

	if r.Header.Get("If-Modified-Since") == auto.AssetsBuildDate {
		w.WriteHeader(http.StatusNotModified)
		return
	}

	mtype := s.mimeTypeForFile(file)
	if len(mtype) != 0 {
		w.Header().Set("Content-Type", mtype)
	}
	if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
		w.Header().Set("Content-Encoding", "gzip")
	} else {
		// ungzip if browser not send gzip accepted header
		var gr *gzip.Reader
		gr, _ = gzip.NewReader(bytes.NewReader(bs))
		bs, _ = ioutil.ReadAll(gr)
		gr.Close()
	}
	w.Header().Set("Content-Length", fmt.Sprintf("%d", len(bs)))
	w.Header().Set("Last-Modified", auto.AssetsBuildDate)
	w.Header().Set("Cache-Control", "public")

	w.Write(bs)
}

func openInvoiceFile(filename string) (io.ReadCloser, func(), error) {
	file, err := os.Open(filename)
	if err != nil {
		return nil, nil, err
	}
	closer := func() { file.Close() }
	var reader io.ReadCloser = file
	var decompressor *gzip.Reader
	if strings.HasSuffix(filename, ".gz") {
		if decompressor, err = gzip.NewReader(file); err != nil {
			return file, closer, err
		}
		closer = func() { decompressor.Close(); file.Close() }
		reader = decompressor
	}
	return reader, closer, nil
}

func readTheThings() {
	file_r_handle, _ := os.OpenFile("testing/testfile.tar.gz", os.O_RDWR, os.ModePerm)
	defer file_r_handle.Close()
	var zip_r_handle *gzip.Reader
	var err error
	if zip_r_handle, err = gzip.NewReader(file_r_handle); err != nil {
		return
	}
	defer zip_r_handle.Close()
	tar_r_handle := tar.NewReader(zip_r_handle)
	tar_r_handle.Next()
	/*
		for header, err := tar_r_handle.Next(); err == nil; tar_r_handle.Next() {
			fmt.Printf("File handle: %s\n", header.Name)
		}
	*/
}

func NewDscin(filename string) *Dscin {
	d := new(Dscin)
	file, err := os.Open(filename)
	if err != nil {
		return nil
	}
	d.closer = func() {
		fmt.Printf("closing file %s\n", filename)
		file.Close()
	}
	var reader io.ReadCloser = file
	var decompressor *gzip.Reader
	if strings.HasSuffix(filename, ".gz") {
		decompressor, _ = gzip.NewReader(file)
		d.closer = func() { decompressor.Close(); file.Close() }
		reader = decompressor
	}
	d.marshaler = formats.GobMarshaler{}
	d.marshaler.ValidateFile(reader)
	return d
}

func main() {
	var file *os.File
	var err error
	var reader *gzip.Reader

	if file, err = os.Open("output/sample.tar.gz"); err != nil {
		log.Fatalln(err)
	}
	defer file.Close()

	if reader, err = gzip.NewReader(file); err != nil {
		log.Fatalln(err)
	}
	defer reader.Close()

	tr := tar.NewReader(reader)

	var header *tar.Header
	for {
		header, err = tr.Next()
		if err == io.EOF {
			// ファイルの最後
			break
		}
		if err != nil {
			log.Fatalln(err)
		}

		buf := new(bytes.Buffer)
		if _, err = io.Copy(buf, tr); err != nil {
			log.Fatalln(err)
		}

		if err = ioutil.WriteFile("output/"+header.Name, buf.Bytes(), 0755); err != nil {
			log.Fatal(err)
		}
	}
}

// deal with .tar.gz
func openTarGz(za *ZipAssets) (err error) {
	var (
		f  *os.File
		tr *tar.Reader
		gr *gzip.Reader
	)

	if f, err = os.Open(za.path); err != nil {
		return
	}
	defer f.Close()

	if gr, err = gzip.NewReader(f); err != nil {
		return
	}
	defer gr.Close()

	tr = tar.NewReader(gr)

	err = openTar(za, tr)

	return
}

// RingOrBuilder attempts to determine whether a file is a Ring or Builder file
// and then loads it accordingly.
func RingOrBuilder(fileName string) (Ring, *Builder, error) {
	var f *os.File
	var r Ring
	var b *Builder
	var err error
	if f, err = os.Open(fileName); err != nil {
		return r, b, err
	}
	var gf *gzip.Reader
	if gf, err = gzip.NewReader(f); err != nil {
		return r, b, err
	}
	header := make([]byte, 16)
	if _, err = io.ReadFull(gf, header); err != nil {
		return r, b, err
	}
	if string(header[:5]) == "RINGv" {
		if string(header[:16]) != RINGVERSION {
			return r, b, fmt.Errorf("Ring Version missmatch, expected %s found %s", RINGVERSION, header[:16])
		}
		gf.Close()
		if _, err = f.Seek(0, 0); err != nil {
			return r, b, err
		}
		r, err = LoadRing(f)
	} else if string(header[:12]) == "RINGBUILDERv" {
		if string(header[:16]) != BUILDERVERSION {
			return r, b, fmt.Errorf("Builder Version missmatch, expected %s found %s", BUILDERVERSION, header[:16])
		}
		gf.Close()
		if _, err = f.Seek(0, 0); err != nil {
			return r, b, err
		}
		b, err = LoadBuilder(f)
	}
	return r, b, err
}

func UnpackTar(filename string, path string) (err error) {
	var file *os.File
	if file, err = os.Open(filename); err != nil {
		return err
	}
	defer file.Close()
	var fileReader io.Reader = file
	var decompressor *gzip.Reader
	if strings.HasSuffix(filename, ".gz") {
		if decompressor, err = gzip.NewReader(file); err != nil {
			return err
		}
		defer decompressor.Close()
	} else if strings.HasSuffix(filename, ".bz2") {
		fileReader = bzip2.NewReader(file)
	}
	var reader *tar.Reader
	if decompressor != nil {
		reader = tar.NewReader(decompressor)
	} else {
		reader = tar.NewReader(fileReader)
	}
	return unpackTarFiles(reader, path)
}

func (hec *HTTPEventCollectorMock) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
	var err error

	hec.numOfRequests++

	if hec.simulateServerError {
		if request.Body != nil {
			defer request.Body.Close()
		}
		writer.WriteHeader(http.StatusInternalServerError)
		return
	}

	switch request.Method {
	case http.MethodOptions:
		// Verify that options method is getting called only once
		if hec.connectionVerified {
			hec.test.Errorf("Connection should not be verified more than once. Got second request with %s method.", request.Method)
		}
		hec.connectionVerified = true
		writer.WriteHeader(http.StatusOK)
	case http.MethodPost:
		// Always verify that Driver is using correct path to HEC
		if request.URL.String() != "/services/collector/event/1.0" {
			hec.test.Errorf("Unexpected path %v", request.URL)
		}
		defer request.Body.Close()

		if authorization, ok := request.Header["Authorization"]; !ok || authorization[0] != ("Splunk "+hec.token) {
			hec.test.Error("Authorization header is invalid.")
		}

		gzipEnabled := false
		if contentEncoding, ok := request.Header["Content-Encoding"]; ok && contentEncoding[0] == "gzip" {
			gzipEnabled = true
		}

		if hec.gzipEnabled == nil {
			hec.gzipEnabled = &gzipEnabled
		} else if gzipEnabled != *hec.gzipEnabled {
			// Nothing wrong with that, but we just know that Splunk Logging Driver does not do that
			hec.test.Error("Driver should not change Content Encoding.")
		}

		var gzipReader *gzip.Reader
		var reader io.Reader
		if gzipEnabled {
			gzipReader, err = gzip.NewReader(request.Body)
			if err != nil {
				hec.test.Fatal(err)
			}
			reader = gzipReader
		} else {
			reader = request.Body
		}

		// Read body
		var body []byte
		body, err = ioutil.ReadAll(reader)
		if err != nil {
			hec.test.Fatal(err)
		}

		// Parse message
		messageStart := 0
		for i := 0; i < len(body); i++ {
			if i == len(body)-1 || (body[i] == '}' && body[i+1] == '{') {
				var message splunkMessage
				err = json.Unmarshal(body[messageStart:i+1], &message)
				if err != nil {
					hec.test.Log(string(body[messageStart : i+1]))
					hec.test.Fatal(err)
				}
				hec.messages = append(hec.messages, &message)
				messageStart = i + 1
			}
		}

		if gzipEnabled {
			gzipReader.Close()
		}

		writer.WriteHeader(http.StatusOK)
	default:
		hec.test.Errorf("Unexpected HTTP method %s", http.MethodOptions)
		writer.WriteHeader(http.StatusBadRequest)
	}
}

func TestGzippedReplication(t *testing.T) {
	if testing.Short() {
		t.Skip()
	}

	cluster := createCluster()

	member1 := createMember(6106)
	repServer1, _ := New(&Config{
		Registrator: cluster.Registrator(member1),
		Membership:  cluster.Membership()})
	defer repServer1.Stop()
	<-repServer1.Sync(1)

	member2 := createMember(6206)
	repServer2, _ := New(&Config{
		Registrator: cluster.Registrator(member2),
		Membership:  cluster.Membership()})
	defer repServer2.Stop()
	<-repServer2.Sync(1)

	var resp *http.Response

	replicator2, _ := repServer2.GetReplicator(auth.NamespaceFrom("ns1"))

	client := http.Client{Transport: &http.Transport{MaxIdleConnsPerHost: 1}}

	var requestData bytes.Buffer
	for i := 0; i < 100; i++ {
		requestData.Write([]byte("9999999999999999999999999999999999"))
	}

	// Iterate over both sync and replication HTTP endpoints
	for _, serviceType := range []string{syncContext, repContext} {
		// Iterate over both content-encoding options- gzip and no zip (identity)
		for _, encoding := range []string{"gzip", "identity"} {
			var dec decoder
			var unZipper *gzip.Reader
			var respBuff bytes.Buffer

			// prepare event that triggers a replication or a sync
			if serviceType == repContext {
				time.AfterFunc(time.Duration(2)*time.Second, func() {
					replicator2.Broadcast(requestData.Bytes())
				})
			} else {
				// grab the channel that was created for us
				// insert an entity inside it and close the channel to signal sync finish
				outMsg := &outMessage{Data: requestData.Bytes()}
				buff, _ := json.Marshal(outMsg)
				time.AfterFunc(time.Duration(2)*time.Second, func() {
					ch := <-repServer2.SyncRequest()
					ch <- buff
					close(ch)
				})
			}

			url := fmt.Sprintf("http://%s:%d/%s/%s", network.GetPrivateIP(), 6206, version, serviceType)
			req, _ := http.NewRequest("GET", url, nil)
			req.Header.Set("Accept-Encoding", encoding)
			req.Header.Set(headerMemberID, "fake member id")
			resp, _ = client.Do(req)
			// wait 4 sec to read enough data, since ioutil.ReadAll blocks until EOF
			// and we use an indefinite connection in case of replication service endpoint
			if serviceType == repContext {
				time.AfterFunc(time.Duration(4)*time.Second, func() { resp.Body.Close() })
			}
			responseData, _ := ioutil.ReadAll(resp.Body)
			respBuff.Write(responseData)

			// if we've sent a gzipped entity, decode it before comparing
			// but also ensure that the returned payload was compressed
			// by comparing the size of the payload to the size sent
			if encoding == "gzip" {
				assert.True(t, len(requestData.Bytes()) > len(responseData),
					"Seems like the gzip hasn't shrunk the payload size enough or at all")
				unZipper, _ = gzip.NewReader(&respBuff)
				dec.Reader = bufio.NewReader(unZipper)
			} else {
				dec.Reader = bufio.NewReader(&respBuff)
			}

			event, _ := dec.Decode()
			if encoding == "gzip" {
				unZipper.Close()
			}
			var entityFromServer outMessage
			json.Unmarshal([]byte(event.Data()), &entityFromServer)
			assert.Equal(t, string(requestData.Bytes()), string(entityFromServer.Data),
				"Didn't receive the same content we sent in the case where encoding is %s", encoding)
		}
	}
}

func releaseGzipReader(zr *gzip.Reader) {
	zr.Close()
	gzipReaderPool.Put(zr)
}