/*
	通過io.Copy實(shí)現(xiàn)可中斷的流復(fù)制
*/
var (
	ak       = "99999999999999999999"
	sk       = "9999999999999999999999999999999999999999"
	endpoint = "http://xx.xx.xx.xx:8060"
	bucket   = "test-bucket"
	key      = "d_xp/2G/2G.txt"
)

func main() {
	s3Client := osg.Client.GetS3Client(ak, sk, endpoint)
	ctx, cancelFunc := context.WithCancel(context.Background())
	object, err := s3Client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
	go func() {
		time.Sleep(time.Second * 10)
		cancelFunc()
		log.Infof("canceled...")
	}()
	if err != nil {
		log.Errorf("%v", err)
		return
	}
	body := object.Body
	defer body.Close()
	file, err := os.Create("/Users/ziyi/GolandProjects/MyTest/demo_home/io_demo/target.txt")
	if err != nil {
		log.Errorf("%v", err)
		return
	}
	defer file.Close()
	_, err = FileService.Copy(ctx, file, body)
	if err != nil {
		log.Errorf("%v", err)
		return
	}

}

type fileService struct {
	sem *semaphore.Weighted
}

var FileService = &fileService{
	sem: semaphore.NewWeighted(1),
}

type IoCopyCancelledErr struct {
	errMsg string
}

func (e *IoCopyCancelledErr) Error() string {
	return fmt.Sprintf("io copy error, %s", e.errMsg)
}

func NewIoCopyCancelledErr(msg string) *IoCopyCancelledErr {
	return &IoCopyCancelledErr{
		errMsg: msg,
	}
}

type readerFunc func(p []byte) (n int, err error)

func (rf readerFunc) Read(p []byte) (n int, err error) { return rf(p) }

//通過ctx實(shí)現(xiàn)可中斷的流拷貝
// Copy closable copy
func (s *fileService) Copy(ctx context.Context, dst io.Writer, src io.Reader) (int64, error) {
	// Copy will call the Reader and Writer interface multiple time, in order
	// to copy by chunk (avoiding loading the whole file in memory).
	// I insert the ability to cancel before read time as it is the earliest
	// possible in the call process.
	size, err := io.Copy(dst, readerFunc(func(p []byte) (int, error) {
		select {
		// if context has been canceled
		case <-ctx.Done():
			// stop process and propagate "context canceled" error
			return 0, NewIoCopyCancelledErr(ctx.Err().Error())
		default:
			// otherwise just run default io.Reader implementation
			return src.Read(p)
		}
	}))
	return size, err
}