JCS-pub/common/pkgs/ec/block.go

package ec

import (
	"errors"
	"io"
	"io/ioutil"

	"gitlink.org.cn/cloudream/common/pkgs/ipfs"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
)

type BlockReader struct {
	ipfsCli *ipfs.PoolClient
	/*将文件分块相关的属性*/
	//fileHash
	fileHash string
	//fileSize
	fileSize int64
	//ecK将文件的分块数
	ecK int
	//chunkSize
	chunkSize int64

	/*可选项*/
	//fastRead,true的时候直接通过hash读block
	jumpReadOpt bool
}

func NewBlockReader() (*BlockReader, error) {
	ipfsClient, err := stgglb.IPFSPool.Acquire()
	if err != nil {
		return nil, err
	}
	//default:fast模式，通过hash直接获取
	return &BlockReader{ipfsCli: ipfsClient, chunkSize: 256 * 1024, jumpReadOpt: false}, nil
}

func (r *BlockReader) Close() {
	r.ipfsCli.Close()
}

func (r *BlockReader) SetJumpRead(fileHash string, fileSize int64, ecK int) {
	r.fileHash = fileHash
	r.fileSize = fileSize
	r.ecK = ecK
	r.jumpReadOpt = true
}

func (r *BlockReader) SetchunkSize(size int64) {
	r.chunkSize = size
}

func (r *BlockReader) FetchBLock(blockHash string) (io.ReadCloser, error) {
	return r.ipfsCli.OpenRead(blockHash)
}

func (r *BlockReader) FetchBLocks(blockHashs []string) ([]io.ReadCloser, error) {
	readers := make([]io.ReadCloser, len(blockHashs))
	for i, hash := range blockHashs {
		var err error
		readers[i], err = r.ipfsCli.OpenRead(hash)
		if err != nil {
			return nil, err
		}
	}
	return readers, nil
}

func (r *BlockReader) JumpFetchBlock(innerID int) (io.ReadCloser, error) {
	if !r.jumpReadOpt {
		return nil, nil
	}
	pipeReader, pipeWriter := io.Pipe()
	go func() {
		for i := int64(r.chunkSize * int64(innerID)); i < r.fileSize; i += int64(r.ecK) * r.chunkSize {
			reader, err := r.ipfsCli.OpenRead(r.fileHash, ipfs.ReadOption{Offset: i, Length: r.chunkSize})
			if err != nil {
				pipeWriter.CloseWithError(err)
				return
			}
			data, err := ioutil.ReadAll(reader)
			if err != nil {
				pipeWriter.CloseWithError(err)
				return
			}
			reader.Close()
			_, err = pipeWriter.Write(data)
			if err != nil {
				pipeWriter.CloseWithError(err)
				return
			}
		}
		//如果文件大小不是分块的整数倍,可能需要补0
		if r.fileSize%(r.chunkSize*int64(r.ecK)) != 0 {
			//pktNum_1:chunkNum-1
			pktNum_1 := r.fileSize / (r.chunkSize * int64(r.ecK))
			offset := (r.fileSize - int64(pktNum_1)*int64(r.ecK)*r.chunkSize)
			count0 := int64(innerID)*int64(r.ecK)*r.chunkSize - offset
			if count0 > 0 {
				add0 := make([]byte, count0)
				pipeWriter.Write(add0)
			}
		}
		pipeWriter.Close()
	}()
	return pipeReader, nil
}

// FetchBlock1这个函数废弃了
func (r *BlockReader) FetchBlock1(input interface{}, errMsg chan error) (io.ReadCloser, error) {
	/*两种模式下传入第一个参数，但是input的类型不同：
	jumpReadOpt-》true：传入blcokHash, string型，通过哈希直接读
	jumpReadOpt->false: 传入innerID，int型，选择需要获取的数据块的id
	*/
	var innerID int
	var blockHash string
	switch input.(type) {
	case int:
		// 执行针对整数的逻辑分支
		if r.jumpReadOpt {
			return nil, errors.New("conflict, wrong input type and jumpReadOpt:true")
		} else {
			innerID = input.(int)
		}
	case string:
		if !r.jumpReadOpt {
			return nil, errors.New("conflict, wrong input type and jumpReadOpt:false")
		} else {
			blockHash = input.(string)
		}
	default:
		return nil, errors.New("wrong input type")
	}
	//开始执行
	if r.jumpReadOpt { //快速读
		ipfsCli, err := stgglb.IPFSPool.Acquire()
		if err != nil {
			logger.Warnf("new ipfs client: %s", err.Error())
			return nil, err
		}
		defer ipfsCli.Close()
		return ipfsCli.OpenRead(blockHash)
	} else { //跳跃读
		ipfsCli, err := stgglb.IPFSPool.Acquire()
		if err != nil {
			logger.Warnf("new ipfs client: %s", err.Error())
			return nil, err
		}
		defer ipfsCli.Close()
		pipeReader, pipeWriter := io.Pipe()
		go func() {
			for i := int64(r.chunkSize * int64(innerID)); i < r.fileSize; i += int64(r.ecK) * r.chunkSize {
				reader, err := ipfsCli.OpenRead(r.fileHash, ipfs.ReadOption{i, r.chunkSize})
				if err != nil {
					pipeWriter.Close()
					errMsg <- err
					return
				}
				data, err := ioutil.ReadAll(reader)
				if err != nil {
					pipeWriter.Close()
					errMsg <- err
					return
				}
				reader.Close()
				_, err = pipeWriter.Write(data)
				if err != nil {
					pipeWriter.Close()
					errMsg <- err
					return
				}
			}
			//如果文件大小不是分块的整数倍,可能需要补0
			if r.fileSize%(r.chunkSize*int64(r.ecK)) != 0 {
				//pktNum_1:chunkNum-1
				pktNum_1 := r.fileSize / (r.chunkSize * int64(r.ecK))
				offset := (r.fileSize - int64(pktNum_1)*int64(r.ecK)*r.chunkSize)
				count0 := int64(innerID)*int64(r.ecK)*r.chunkSize - offset
				if count0 > 0 {
					add0 := make([]byte, count0)
					pipeWriter.Write(add0)
				}
			}
			pipeWriter.Close()
			errMsg <- nil
		}()
		return pipeReader, nil
	}
}