--- /dev/null
+package oss
+
+import (
+ "bytes"
+ "encoding/binary"
+ "fmt"
+ "hash"
+ "hash/crc32"
+ "io"
+ "net/http"
+ "time"
+)
+
+// The adapter class for Select object's response.
+// The response consists of frames. Each frame has the following format:
+
+// Type | Payload Length | Header Checksum | Payload | Payload Checksum
+
+// |<4-->| <--4 bytes------><---4 bytes-------><-n/a-----><--4 bytes--------->
+// And we have three kind of frames.
+// Data Frame:
+// Type:8388609
+// Payload: Offset | Data
+// <-8 bytes>
+
+// Continuous Frame
+// Type:8388612
+// Payload: Offset (8-bytes)
+
+// End Frame
+// Type:8388613
+// Payload: Offset | total scanned bytes | http status code | error message
+// <-- 8bytes--><-----8 bytes--------><---4 bytes-------><---variabe--->
+
+// SelectObjectResponse defines HTTP response from OSS SelectObject
+type SelectObjectResponse struct {
+ StatusCode int
+ Headers http.Header
+ Body io.ReadCloser
+ Frame SelectObjectResult
+ ReadTimeOut uint
+ ClientCRC32 uint32
+ ServerCRC32 uint32
+ WriterForCheckCrc32 hash.Hash32
+ Finish bool
+}
+
+func (sr *SelectObjectResponse) Read(p []byte) (n int, err error) {
+ n, err = sr.readFrames(p)
+ return
+}
+
+// Close http reponse body
+func (sr *SelectObjectResponse) Close() error {
+ return sr.Body.Close()
+}
+
+// PostSelectResult is the request of SelectObject
+type PostSelectResult struct {
+ Response *SelectObjectResponse
+}
+
+// readFrames is read Frame
+func (sr *SelectObjectResponse) readFrames(p []byte) (int, error) {
+ var nn int
+ var err error
+ var checkValid bool
+ if sr.Frame.OutputRawData == true {
+ nn, err = sr.Body.Read(p)
+ return nn, err
+ }
+
+ if sr.Finish {
+ return 0, io.EOF
+ }
+
+ for {
+ // if this Frame is Readed, then not reading Header
+ if sr.Frame.OpenLine != true {
+ err = sr.analysisHeader()
+ if err != nil {
+ return nn, err
+ }
+ }
+
+ if sr.Frame.FrameType == DataFrameType {
+ n, err := sr.analysisData(p[nn:])
+ if err != nil {
+ return nn, err
+ }
+ nn += n
+
+ // if this Frame is readed all data, then empty the Frame to read it with next frame
+ if sr.Frame.ConsumedBytesLength == sr.Frame.PayloadLength-8 {
+ checkValid, err = sr.checkPayloadSum()
+ if err != nil || !checkValid {
+ return nn, fmt.Errorf("%s", err.Error())
+ }
+ sr.emptyFrame()
+ }
+
+ if nn == len(p) {
+ return nn, nil
+ }
+ } else if sr.Frame.FrameType == ContinuousFrameType {
+ checkValid, err = sr.checkPayloadSum()
+ if err != nil || !checkValid {
+ return nn, fmt.Errorf("%s", err.Error())
+ }
+ } else if sr.Frame.FrameType == EndFrameType {
+ err = sr.analysisEndFrame()
+ if err != nil {
+ return nn, err
+ }
+ checkValid, err = sr.checkPayloadSum()
+ if checkValid {
+ sr.Finish = true
+ }
+ return nn, err
+ } else if sr.Frame.FrameType == MetaEndFrameCSVType {
+ err = sr.analysisMetaEndFrameCSV()
+ if err != nil {
+ return nn, err
+ }
+ checkValid, err = sr.checkPayloadSum()
+ if checkValid {
+ sr.Finish = true
+ }
+ return nn, err
+ } else if sr.Frame.FrameType == MetaEndFrameJSONType {
+ err = sr.analysisMetaEndFrameJSON()
+ if err != nil {
+ return nn, err
+ }
+ checkValid, err = sr.checkPayloadSum()
+ if checkValid {
+ sr.Finish = true
+ }
+ return nn, err
+ }
+ }
+ return nn, nil
+}
+
+type chanReadIO struct {
+ readLen int
+ err error
+}
+
+func (sr *SelectObjectResponse) readLen(p []byte, timeOut time.Duration) (int, error) {
+ r := sr.Body
+ ch := make(chan chanReadIO, 1)
+ defer close(ch)
+ go func(p []byte) {
+ var needReadLength int
+ readChan := chanReadIO{}
+ needReadLength = len(p)
+ for {
+ n, err := r.Read(p[readChan.readLen:needReadLength])
+ readChan.readLen += n
+ if err != nil {
+ readChan.err = err
+ ch <- readChan
+ return
+ }
+
+ if readChan.readLen == needReadLength {
+ break
+ }
+ }
+ ch <- readChan
+ }(p)
+
+ select {
+ case <-time.After(time.Second * timeOut):
+ return 0, fmt.Errorf("requestId: %s, readLen timeout, timeout is %d(second),need read:%d", sr.Headers.Get(HTTPHeaderOssRequestID), timeOut, len(p))
+ case result := <-ch:
+ return result.readLen, result.err
+ }
+}
+
+// analysisHeader is reading selectObject response body's header
+func (sr *SelectObjectResponse) analysisHeader() error {
+ headFrameByte := make([]byte, 20)
+ _, err := sr.readLen(headFrameByte, time.Duration(sr.ReadTimeOut))
+ if err != nil {
+ return fmt.Errorf("requestId: %s, Read response frame header failure,err:%s", sr.Headers.Get(HTTPHeaderOssRequestID), err.Error())
+ }
+
+ frameTypeByte := headFrameByte[0:4]
+ sr.Frame.Version = frameTypeByte[0]
+ frameTypeByte[0] = 0
+ bytesToInt(frameTypeByte, &sr.Frame.FrameType)
+
+ if sr.Frame.FrameType != DataFrameType && sr.Frame.FrameType != ContinuousFrameType &&
+ sr.Frame.FrameType != EndFrameType && sr.Frame.FrameType != MetaEndFrameCSVType && sr.Frame.FrameType != MetaEndFrameJSONType {
+ return fmt.Errorf("requestId: %s, Unexpected frame type: %d", sr.Headers.Get(HTTPHeaderOssRequestID), sr.Frame.FrameType)
+ }
+
+ payloadLengthByte := headFrameByte[4:8]
+ bytesToInt(payloadLengthByte, &sr.Frame.PayloadLength)
+ headCheckSumByte := headFrameByte[8:12]
+ bytesToInt(headCheckSumByte, &sr.Frame.HeaderCheckSum)
+ byteOffset := headFrameByte[12:20]
+ bytesToInt(byteOffset, &sr.Frame.Offset)
+ sr.Frame.OpenLine = true
+
+ err = sr.writerCheckCrc32(byteOffset)
+ return err
+}
+
+// analysisData is reading the DataFrameType data of selectObject response body
+func (sr *SelectObjectResponse) analysisData(p []byte) (int, error) {
+ var needReadLength int32
+ lenP := int32(len(p))
+ restByteLength := sr.Frame.PayloadLength - 8 - sr.Frame.ConsumedBytesLength
+ if lenP <= restByteLength {
+ needReadLength = lenP
+ } else {
+ needReadLength = restByteLength
+ }
+ n, err := sr.readLen(p[:needReadLength], time.Duration(sr.ReadTimeOut))
+ if err != nil {
+ return n, fmt.Errorf("read frame data error,%s", err.Error())
+ }
+ sr.Frame.ConsumedBytesLength += int32(n)
+ err = sr.writerCheckCrc32(p[:n])
+ return n, err
+}
+
+// analysisEndFrame is reading the EndFrameType data of selectObject response body
+func (sr *SelectObjectResponse) analysisEndFrame() error {
+ var eF EndFrame
+ payLoadBytes := make([]byte, sr.Frame.PayloadLength-8)
+ _, err := sr.readLen(payLoadBytes, time.Duration(sr.ReadTimeOut))
+ if err != nil {
+ return fmt.Errorf("read end frame error:%s", err.Error())
+ }
+ bytesToInt(payLoadBytes[0:8], &eF.TotalScanned)
+ bytesToInt(payLoadBytes[8:12], &eF.HTTPStatusCode)
+ errMsgLength := sr.Frame.PayloadLength - 20
+ eF.ErrorMsg = string(payLoadBytes[12 : errMsgLength+12])
+ sr.Frame.EndFrame.TotalScanned = eF.TotalScanned
+ sr.Frame.EndFrame.HTTPStatusCode = eF.HTTPStatusCode
+ sr.Frame.EndFrame.ErrorMsg = eF.ErrorMsg
+ err = sr.writerCheckCrc32(payLoadBytes)
+ return err
+}
+
+// analysisMetaEndFrameCSV is reading the MetaEndFrameCSVType data of selectObject response body
+func (sr *SelectObjectResponse) analysisMetaEndFrameCSV() error {
+ var mCF MetaEndFrameCSV
+ payLoadBytes := make([]byte, sr.Frame.PayloadLength-8)
+ _, err := sr.readLen(payLoadBytes, time.Duration(sr.ReadTimeOut))
+ if err != nil {
+ return fmt.Errorf("read meta end csv frame error:%s", err.Error())
+ }
+
+ bytesToInt(payLoadBytes[0:8], &mCF.TotalScanned)
+ bytesToInt(payLoadBytes[8:12], &mCF.Status)
+ bytesToInt(payLoadBytes[12:16], &mCF.SplitsCount)
+ bytesToInt(payLoadBytes[16:24], &mCF.RowsCount)
+ bytesToInt(payLoadBytes[24:28], &mCF.ColumnsCount)
+ errMsgLength := sr.Frame.PayloadLength - 36
+ mCF.ErrorMsg = string(payLoadBytes[28 : errMsgLength+28])
+ sr.Frame.MetaEndFrameCSV.ErrorMsg = mCF.ErrorMsg
+ sr.Frame.MetaEndFrameCSV.TotalScanned = mCF.TotalScanned
+ sr.Frame.MetaEndFrameCSV.Status = mCF.Status
+ sr.Frame.MetaEndFrameCSV.SplitsCount = mCF.SplitsCount
+ sr.Frame.MetaEndFrameCSV.RowsCount = mCF.RowsCount
+ sr.Frame.MetaEndFrameCSV.ColumnsCount = mCF.ColumnsCount
+ err = sr.writerCheckCrc32(payLoadBytes)
+ return err
+}
+
+// analysisMetaEndFrameJSON is reading the MetaEndFrameJSONType data of selectObject response body
+func (sr *SelectObjectResponse) analysisMetaEndFrameJSON() error {
+ var mJF MetaEndFrameJSON
+ payLoadBytes := make([]byte, sr.Frame.PayloadLength-8)
+ _, err := sr.readLen(payLoadBytes, time.Duration(sr.ReadTimeOut))
+ if err != nil {
+ return fmt.Errorf("read meta end json frame error:%s", err.Error())
+ }
+
+ bytesToInt(payLoadBytes[0:8], &mJF.TotalScanned)
+ bytesToInt(payLoadBytes[8:12], &mJF.Status)
+ bytesToInt(payLoadBytes[12:16], &mJF.SplitsCount)
+ bytesToInt(payLoadBytes[16:24], &mJF.RowsCount)
+ errMsgLength := sr.Frame.PayloadLength - 32
+ mJF.ErrorMsg = string(payLoadBytes[24 : errMsgLength+24])
+ sr.Frame.MetaEndFrameJSON.ErrorMsg = mJF.ErrorMsg
+ sr.Frame.MetaEndFrameJSON.TotalScanned = mJF.TotalScanned
+ sr.Frame.MetaEndFrameJSON.Status = mJF.Status
+ sr.Frame.MetaEndFrameJSON.SplitsCount = mJF.SplitsCount
+ sr.Frame.MetaEndFrameJSON.RowsCount = mJF.RowsCount
+
+ err = sr.writerCheckCrc32(payLoadBytes)
+ return err
+}
+
+func (sr *SelectObjectResponse) checkPayloadSum() (bool, error) {
+ payLoadChecksumByte := make([]byte, 4)
+ n, err := sr.readLen(payLoadChecksumByte, time.Duration(sr.ReadTimeOut))
+ if n == 4 {
+ bytesToInt(payLoadChecksumByte, &sr.Frame.PayloadChecksum)
+ sr.ServerCRC32 = sr.Frame.PayloadChecksum
+ sr.ClientCRC32 = sr.WriterForCheckCrc32.Sum32()
+ if sr.Frame.EnablePayloadCrc == true && sr.ServerCRC32 != 0 && sr.ServerCRC32 != sr.ClientCRC32 {
+ return false, fmt.Errorf("RequestId: %s, Unexpected frame type: %d, client %d but server %d",
+ sr.Headers.Get(HTTPHeaderOssRequestID), sr.Frame.FrameType, sr.ClientCRC32, sr.ServerCRC32)
+ }
+ return true, err
+ }
+ return false, fmt.Errorf("RequestId:%s, read checksum error:%s", sr.Headers.Get(HTTPHeaderOssRequestID), err.Error())
+}
+
+func (sr *SelectObjectResponse) writerCheckCrc32(p []byte) (err error) {
+ err = nil
+ if sr.Frame.EnablePayloadCrc == true {
+ _, err = sr.WriterForCheckCrc32.Write(p)
+ }
+ return err
+}
+
+// emptyFrame is emptying SelectObjectResponse Frame information
+func (sr *SelectObjectResponse) emptyFrame() {
+ crcCalc := crc32.NewIEEE()
+ sr.WriterForCheckCrc32 = crcCalc
+ sr.Finish = false
+
+ sr.Frame.ConsumedBytesLength = 0
+ sr.Frame.OpenLine = false
+ sr.Frame.Version = byte(0)
+ sr.Frame.FrameType = 0
+ sr.Frame.PayloadLength = 0
+ sr.Frame.HeaderCheckSum = 0
+ sr.Frame.Offset = 0
+ sr.Frame.Data = ""
+
+ sr.Frame.EndFrame.TotalScanned = 0
+ sr.Frame.EndFrame.HTTPStatusCode = 0
+ sr.Frame.EndFrame.ErrorMsg = ""
+
+ sr.Frame.MetaEndFrameCSV.TotalScanned = 0
+ sr.Frame.MetaEndFrameCSV.Status = 0
+ sr.Frame.MetaEndFrameCSV.SplitsCount = 0
+ sr.Frame.MetaEndFrameCSV.RowsCount = 0
+ sr.Frame.MetaEndFrameCSV.ColumnsCount = 0
+ sr.Frame.MetaEndFrameCSV.ErrorMsg = ""
+
+ sr.Frame.MetaEndFrameJSON.TotalScanned = 0
+ sr.Frame.MetaEndFrameJSON.Status = 0
+ sr.Frame.MetaEndFrameJSON.SplitsCount = 0
+ sr.Frame.MetaEndFrameJSON.RowsCount = 0
+ sr.Frame.MetaEndFrameJSON.ErrorMsg = ""
+
+ sr.Frame.PayloadChecksum = 0
+}
+
+// bytesToInt byte's array trans to int
+func bytesToInt(b []byte, ret interface{}) {
+ binBuf := bytes.NewBuffer(b)
+ binary.Read(binBuf, binary.BigEndian, ret)
+}
OSDN Git Service
