Repository: dragonflyoss/Dragonfly2
Pull request #877 (2021-12-06 08:18:16)

unify back source interface
<!--- Provide a general summary of your changes in the Title above -->

## Description

unify back source interface
<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate):

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] Code compiles correctly.
- [ ] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.
- [ ] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#877](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (cd39e0f) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/6bbd7e647b44a2e0a8e5e73d4410a71951dc0a53?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (6bbd7e6) will **increase** coverage by `0.12%`.
> The diff coverage is `40.51%`.

> :exclamation: Current head cd39e0f differs from pull request most recent head 2b4c39d. Consider uploading reports for the commit 2b4c39d to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main     #877      +/-   ##
==========================================
+ Coverage   41.14%   41.26%   +0.12%     
==========================================
  Files         102      105       +3     
  Lines        8057     8254     +197     
==========================================
+ Hits         3315     3406      +91     
- Misses       4330     4409      +79     
- Partials      412      439      +27     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `41.26% <40.51%> (+0.12%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [cdn/supervisor/task/manager\_util.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvdGFzay9tYW5hZ2VyX3V0aWwuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/source/plugin.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9wbHVnaW4uZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/source/request\_fixture.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9yZXF1ZXN0X2ZpeHR1cmUuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/source/source\_client.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9zb3VyY2VfY2xpZW50Lmdv) | `1.92% <2.27%> (-0.12%)` | :arrow_down: |
| [client/dfget/dfget.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RmZ2V0L2RmZ2V0Lmdv) | `15.74% <25.00%> (+0.42%)` | :arrow_up: |
| [pkg/source/header.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9oZWFkZXIuZ28=) | `42.85% <42.85%> (ø)` | |
| [client/daemon/peer/piece\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BpZWNlX21hbmFnZXIuZ28=) | `62.28% <44.44%> (-0.82%)` | :arrow_down: |
| [cdn/supervisor/cdn/cache\_detector.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvY2RuL2NhY2hlX2RldGVjdG9yLmdv) | `55.72% <50.00%> (-1.54%)` | :arrow_down: |
| [pkg/source/httpprotocol/http\_source\_client.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9odHRwcHJvdG9jb2wvaHR0cF9zb3VyY2VfY2xpZW50Lmdv) | `56.31% <50.00%> (-10.36%)` | :arrow_down: |
| [pkg/source/request.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9yZXF1ZXN0Lmdv) | `50.98% <50.98%> (ø)` | |
| ... and [11 more](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/877/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | |


Patch:
diff --git a/cdn/supervisor/cdn/cache_detector.go b/cdn/supervisor/cdn/cache_detector.go
index 6094394b..fd41309e 100644
--- a/cdn/supervisor/cdn/cache_detector.go
+++ b/cdn/supervisor/cdn/cache_detector.go
@@ -24,7 +24,6 @@ import (
 	"io"
 	"io/ioutil"
 	"sort"
-	"time"
 
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/trace"
@@ -103,16 +102,22 @@ func (cd *cacheDetector) doDetect(ctx context.Context, task *types.SeedTask, fil
 	if err := checkSameFile(task, fileMetaData); err != nil {
 		return nil, errors.Wrapf(err, "check same file")
 	}
-	ctx, expireCancel := context.WithTimeout(context.Background(), 4*time.Second)
-	defer expireCancel()
-	expired, err := source.IsExpired(ctx, task.URL, task.Header, fileMetaData.ExpireInfo)
+	checkExpiredRequest, err := source.NewRequestWithContext(ctx, task.URL, task.Header)
 	if err != nil {
-		// 如果获取失败，则认为没有过期，防止打爆源
-		task.Log().Errorf("failed to check if the task expired: %v", err)
+		return nil, errors.Wrapf(err, "create request")
 	}
-	task.Log().Debugf("task expired result: %t", expired)
+	expired, err := source.IsExpired(checkExpiredRequest, &source.ExpireInfo{
+		LastModified: fileMetaData.ExpireInfo[source.LastModified],
+		ETag:         fileMetaData.ExpireInfo[source.ETag],
+	})
+	if err != nil {
+		// If the check fails, the resource is regarded as not expired to prevent the source from being knocked down
+		task.Log().Warnf("failed to check whether the source is expired. To prevent the source from being suspended, "+
+			"assume that the source is not expired: %v", err)
+	}
+	task.Log().Debugf("task resource expired result: %t", expired)
 	if expired {
-		return nil, cdnerrors.ErrResourceExpired{URL: task.URL}
+		return nil, errors.Errorf("resource %s has expired", task.TaskURL)
 	}
 	// not expired
 	if fileMetaData.Finish {
@@ -121,14 +126,17 @@ func (cd *cacheDetector) doDetect(ctx context.Context, task *types.SeedTask, fil
 	}
 	// check if the resource supports range request. if so,
 	// detect the cache situation by reading piece meta and data file
-	ctx, rangeCancel := context.WithTimeout(context.Background(), 4*time.Second)
-	defer rangeCancel()
-	supportRange, err := source.IsSupportRange(ctx, task.URL, task.Header)
+	checkSupportRangeRequest, err := source.NewRequestWithContext(ctx, task.URL, task.Header)
+	if err != nil {
+		return nil, errors.Wrapf(err, "create check support range request")
+	}
+	checkSupportRangeRequest.Header.Add(source.Range, "0-0")
+	supportRange, err := source.IsSupportRange(checkSupportRangeRequest)
 	if err != nil {
-		return nil, errors.Wrapf(err, "check if url(%s) supports range request", task.URL)
+		return nil, errors.Wrap(err, "check if support range")
 	}
 	if !supportRange {
-		return nil, cdnerrors.ErrResourceNotSupportRangeRequest{URL: task.URL}
+		return nil, errors.Errorf("resource %s is not support range request", task.URL)
 	}
 	return cd.parseByReadFile(task.TaskID, fileMetaData, fileDigest)
 }
@@ -169,12 +177,12 @@ func (cd *cacheDetector) parseByReadMetaFile(taskID string, fileMetaData *storag
 func (cd *cacheDetector) parseByReadFile(taskID string, metaData *storage.FileMetaData, fileDigest hash.Hash) (*cacheResult, error) {
 	reader, err := cd.cacheDataManager.readDownloadFile(taskID)
 	if err != nil {
-		return nil, errors.Wrapf(err, "read data file")
+		return nil, errors.Wrapf(err, "read download data file")
 	}
 	defer reader.Close()
 	tempRecords, err := cd.cacheDataManager.readPieceMetaRecords(taskID)
 	if err != nil {
-		return nil, errors.Wrapf(err, "read piece meta file")
+		return nil, errors.Wrapf(err, "read piece meta records")
 	}
 
 	// sort piece meta records by pieceNum
@@ -188,7 +196,7 @@ func (cd *cacheDetector) parseByReadFile(taskID string, metaData *storage.FileMe
 		if uint32(index) != tempRecords[index].PieceNum {
 			break
 		}
-		// read content
+		// read content TODO concurrent by multi-goroutine
 		if err := checkPieceContent(reader, tempRecords[index], fileDigest); err != nil {
 			logger.WithTaskID(taskID).Errorf("read content of pieceNum %d failed: %v", tempRecords[index].PieceNum, err)
 			break
diff --git a/cdn/supervisor/cdn/downloader.go b/cdn/supervisor/cdn/downloader.go
index c6ce3609..208a6b0c 100644
--- a/cdn/supervisor/cdn/downloader.go
+++ b/cdn/supervisor/cdn/downloader.go
@@ -25,34 +25,55 @@ import (
 
 	"d7y.io/dragonfly/v2/cdn/types"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/util/maputils"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+	"d7y.io/dragonfly/v2/pkg/util/stringutils"
 )
 
-const RangeHeaderName = "Range"
-
-func (cm *Manager) download(ctx context.Context, task *types.SeedTask, detectResult *cacheResult) (io.ReadCloser, error) {
-	headers := maputils.DeepCopy(task.Header)
-	if detectResult.breakPoint > 0 {
-		breakRange, err := rangeutils.GetBreakRange(detectResult.breakPoint, task.SourceFileLength)
+func (cm *Manager) download(ctx context.Context, task *types.SeedTask, breakPoint int64) (io.ReadCloser, error) {
+	var err error
+	breakRange := task.Range
+	if breakPoint > 0 {
+		// todo replace task.SourceFileLength with totalSourceFileLength to get BreakRange
+		breakRange, err = getBreakRange(breakPoint, task.Range, task.SourceFileLength)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to calculate the breakRange")
-		}
-		// check if Range in header? if Range already in Header, priority use this range
-		if _, ok := headers[RangeHeaderName]; !ok {
-			headers[RangeHeaderName] = fmt.Sprintf("bytes=%s", breakRange)
+			return nil, errors.Wrapf(err, "calculate the breakRange")
 		}
 	}
-	task.Log().Infof("start download url %s at range: %d-%d: with header: %#v", task.URL, detectResult.breakPoint,
-		task.SourceFileLength, task.Header)
-	reader, responseHeader, err := source.DownloadWithResponseHeader(ctx, task.URL, headers)
+	task.Log().Infof("start downloading URL %s at range %s with header %s", task.URL, breakRange, task.Header)
+	downloadRequest, err := source.NewRequestWithContext(ctx, task.URL, task.Header)
+	if err != nil {
+		return nil, errors.Wrap(err, "create download request")
+	}
+	if !stringutils.IsBlank(breakRange) {
+		downloadRequest.Header.Add(source.Range, breakRange)
+	}
+	body, expireInfo, err := source.DownloadWithExpireInfo(downloadRequest)
 	// update Expire info
 	if err == nil {
-		expireInfo := map[string]string{
-			source.LastModified: responseHeader.Get(source.LastModified),
-			source.ETag:         responseHeader.Get(source.ETag),
-		}
-		cm.updateExpireInfo(task.TaskID, expireInfo)
+		cm.updateExpireInfo(task.TaskID, map[string]string{
+			source.LastModified: expireInfo.LastModified,
+			source.ETag:         expireInfo.ETag,
+		})
+	}
+	return body, err
+}
+
+func getBreakRange(breakPoint int64, taskRange string, fileTotalLength int64) (string, error) {
+	if breakPoint <= 0 {
+		return "", errors.Errorf("breakPoint is non-positive: %d", breakPoint)
+	}
+	if fileTotalLength <= 0 {
+		return "", errors.Errorf("file length is non-positive: %d", fileTotalLength)
+	}
+	if stringutils.IsBlank(taskRange) {
+		return fmt.Sprintf("%d-%d", breakPoint, fileTotalLength-1), nil
+	}
+	requestRange, err := rangeutils.ParseRange(taskRange, uint64(fileTotalLength))
+	if err != nil {
+		return "", errors.Errorf("parse range failed, taskRange: %s, fileTotalLength: %d: %v", taskRange, fileTotalLength, err)
+	}
+	if breakPoint >= int64(requestRange.EndIndex-requestRange.StartIndex+1) {
+		return "", errors.Errorf("breakPoint %d is larger than or equal with length of download required %s", breakPoint, requestRange)
 	}
-	return reader, err
+	return fmt.Sprintf("%d-%d", requestRange.StartIndex+uint64(breakPoint), requestRange.EndIndex), nil
 }
diff --git a/cdn/supervisor/cdn/manager.go b/cdn/supervisor/cdn/manager.go
index cf24afeb..272122db 100644
--- a/cdn/supervisor/cdn/manager.go
+++ b/cdn/supervisor/cdn/manager.go
@@ -129,7 +129,7 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	var downloadSpan trace.Span
 	ctx, downloadSpan = tracer.Start(ctx, config.SpanDownloadSource)
 	downloadSpan.End()
-	body, err := cm.download(ctx, task, detectResult)
+	body, err := cm.download(ctx, task, detectResult.breakPoint)
 	// download fail
 	if err != nil {
 		downloadSpan.RecordError(err)
diff --git a/cdn/supervisor/cdn/storage/storage_mgr.go b/cdn/supervisor/cdn/storage/storage.go
similarity index 98%
rename from cdn/supervisor/cdn/storage/storage_mgr.go
rename to cdn/supervisor/cdn/storage/storage.go
index 960ac2d8..ddd9271f 100644
--- a/cdn/supervisor/cdn/storage/storage_mgr.go
+++ b/cdn/supervisor/cdn/storage/storage.go
@@ -128,11 +128,11 @@ func ParsePieceMetaRecord(value string) (record *PieceMetaRecord, err error) {
 		return nil, errors.Wrapf(err, "invalid pieceLen: %s", fields[1])
 	}
 	md5 := fields[2]
-	pieceRange, err := rangeutils.ParseRange(fields[3])
+	pieceRange, err := rangeutils.GetRange(fields[3])
 	if err != nil {
 		return nil, errors.Wrapf(err, "invalid piece range: %s", fields[3])
 	}
-	originRange, err := rangeutils.ParseRange(fields[4])
+	originRange, err := rangeutils.GetRange(fields[4])
 	if err != nil {
 		return nil, errors.Wrapf(err, "invalid origin range: %s", fields[4])
 	}
diff --git a/cdn/supervisor/mock/mock_task_mgr.go b/cdn/supervisor/mock/mock_task_mgr.go
index 070c3ebe..fc83c12b 100644
--- a/cdn/supervisor/mock/mock_task_mgr.go
+++ b/cdn/supervisor/mock/mock_task_mgr.go
@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: d7y.io/dragonfly/v2/cdnsystem/supervisor (interfaces: SeedTaskMgr)
+// Source: d7y.io/dragonfly/v2/cdn/supervisor (interfaces: SeedTaskMgr)
 
 // Package mock is a generated GoMock package.
 package mock
diff --git a/cdn/supervisor/task/manager_util.go b/cdn/supervisor/task/manager_util.go
index 02494eb0..ff6b7f21 100644
--- a/cdn/supervisor/task/manager_util.go
+++ b/cdn/supervisor/task/manager_util.go
@@ -85,11 +85,13 @@ func (tm *Manager) addOrUpdateTask(ctx context.Context, request *types.TaskRegis
 		return task, nil
 	}
 
+	lengthRequest, err := source.NewRequestWithContext(ctx, task.URL, request.Header)
+	if err != nil {
+		return nil, err
+	}
 	// get sourceContentLength with req.Header
-	ctx, cancel := context.WithTimeout(ctx, 4*time.Second)
-	defer cancel()
 	span.AddEvent(config.EventRequestSourceFileLength)
-	sourceFileLength, err := source.GetContentLength(ctx, task.URL, request.Header)
+	sourceFileLength, err := source.GetContentLength(lengthRequest)
 	if err != nil {
 		task.Log().Errorf("failed to get url (%s) content length: %v", task.URL, err)
 		if cdnerrors.IsURLNotReachable(err) {
diff --git a/cdn/types/seed_task_info.go b/cdn/types/seed_task_info.go
index b7b14024..4a131683 100644
--- a/cdn/types/seed_task_info.go
+++ b/cdn/types/seed_task_info.go
@@ -31,6 +31,7 @@ type SeedTask struct {
 	CdnStatus        string            `json:"cdnStatus,omitempty"`
 	PieceTotal       int32             `json:"pieceTotal,omitempty"`
 	RequestDigest    string            `json:"requestDigest,omitempty"`
+	Range            string            `json:"range"`
 	SourceRealDigest string            `json:"sourceRealDigest,omitempty"`
 	PieceMd5Sign     string            `json:"pieceMd5Sign,omitempty"`
 	logger           *logger.SugaredLoggerOnWith
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index 542dfd29..06c1a475 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -347,7 +347,11 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 	}
 	log := pt.Log()
 	log.Infof("start to download from source")
-	contentLength, err := source.GetContentLength(ctx, request.Url, request.UrlMeta.Header)
+	contentLengthRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
+	if err != nil {
+		return err
+	}
+	contentLength, err := source.GetContentLength(contentLengthRequest)
 	if err != nil {
 		log.Warnf("get content length error: %s for %s", err, request.Url)
 	}
@@ -368,7 +372,11 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 	}
 	log.Debugf("get content length: %d", contentLength)
 	// 1. download piece from source
-	body, err := source.Download(ctx, request.Url, request.UrlMeta.Header)
+	downloadRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
+	if err != nil {
+		return err
+	}
+	body, err := source.Download(downloadRequest)
 	if err != nil {
 		return err
 	}
diff --git a/client/dfget/dfget.go b/client/dfget/dfget.go
index 6bfb5cec..cebfd964 100644
--- a/client/dfget/dfget.go
+++ b/client/dfget/dfget.go
@@ -148,7 +148,11 @@ func downloadFromSource(ctx context.Context, cfg *config.DfgetConfig, hdr map[st
 	defer os.Remove(target.Name())
 	defer target.Close()
 
-	if response, err = source.Download(ctx, cfg.URL, hdr); err != nil {
+	downloadRequest, err := source.NewRequestWithContext(ctx, cfg.URL, hdr)
+	if err != nil {
+		return err
+	}
+	if response, err = source.Download(downloadRequest); err != nil {
 		return err
 	}
 	defer response.Close()
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client.go b/pkg/source/hdfsprotocol/hdfs_source_client.go
index bd0a619d..b0c0e1f3 100644
--- a/pkg/source/hdfsprotocol/hdfs_source_client.go
+++ b/pkg/source/hdfsprotocol/hdfs_source_client.go
@@ -18,7 +18,6 @@ package hdfsprotocol
 
 import (
 	"bytes"
-	"context"
 	"io"
 	"net/url"
 	"os/user"
@@ -27,18 +26,17 @@ import (
 	"time"
 
 	"github.com/colinmarc/hdfs/v2"
-	"github.com/go-http-utils/headers"
 	"github.com/pkg/errors"
 
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 )
 
 const (
 	HDFSClient = "hdfs"
 )
 const (
-	layout = "2006-01-02 15:04:05"
 	// hdfsUseDataNodeHostName set hdfs client whether user hostname connect to datanode
 	hdfsUseDataNodeHostName = "dfs.client.use.datanode.hostname"
 	// hdfsUseDataNodeHostNameValue set value is true
@@ -46,7 +44,14 @@ const (
 )
 
 func init() {
-	source.Register(HDFSClient, NewHDFSSourceClient())
+	if err := source.Register(HDFSClient, NewHDFSSourceClient(), adapter); err != nil {
+		panic(err)
+	}
+}
+
+func adapter(request *source.Request) *source.Request {
+	clonedRequest := request.Clone(request.Context())
+	return clonedRequest
 }
 
 // hdfsSourceClient is an implementation of the interface of SourceClient.
@@ -55,7 +60,7 @@ type hdfsSourceClient struct {
 	clientMap map[string]*hdfs.Client
 }
 
-// hdfsFileReaderClose is an combination object of the  io.LimitedReader and io.Closer
+// hdfsFileReaderClose is a combination object of the  io.LimitedReader and io.Closer
 type hdfsFileReaderClose struct {
 	limited io.Reader
 	c       io.Closer
@@ -72,29 +77,20 @@ func newHdfsFileReaderClose(r io.Reader, n int64, c io.Closer) io.ReadCloser {
 
 type HDFSSourceClientOption func(p *hdfsSourceClient)
 
-func (h *hdfsSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+func (h *hdfsSourceClient) GetContentLength(request *source.Request) (int64, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
-		return -1, err
+		return source.UnKnownSourceFileLen, err
 	}
 	info, err := hdfsClient.Stat(path)
 	if err != nil {
-		return -1, err
+		return source.UnKnownSourceFileLen, err
 	}
-
-	if rang != nil {
-
-		if int64(rang.EndIndex) <= info.Size() {
-			return int64(rang.EndIndex - rang.StartIndex), nil
-		}
-		return info.Size() - int64(rang.StartIndex), nil
-	}
-
 	return info.Size(), nil
 }
 
-func (h *hdfsSourceClient) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+func (h *hdfsSourceClient) IsSupportRange(request *source.Request) (bool, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
 		return false, err
 	}
@@ -105,31 +101,21 @@ func (h *hdfsSourceClient) IsSupportRange(ctx context.Context, url string, heade
 	return true, nil
 }
 
-func (h *hdfsSourceClient) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
-	lastModified := expireInfo[headers.LastModified]
-	//eTag := expireInfo[headers.ETag]
-	if lastModified == "" {
-		return true, nil
-	}
-
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+func (h *hdfsSourceClient) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
 		return false, err
 	}
 
-	info, err := hdfsClient.Stat(path)
-	if err != nil {
-		return false, err
-	}
-	t, err := time.ParseInLocation(layout, lastModified, time.Local)
+	fileInfo, err := hdfsClient.Stat(path)
 	if err != nil {
 		return false, err
 	}
-	return info.ModTime().Format(layout) != t.Format(layout), nil
+	return fileInfo.ModTime().Format(source.LastModifiedLayout) != info.LastModified, nil
 }
 
-func (h *hdfsSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+func (h *hdfsSourceClient) Download(request *source.Request) (io.ReadCloser, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
 		return nil, err
 	}
@@ -139,23 +125,29 @@ func (h *hdfsSourceClient) Download(ctx context.Context, url string, header sour
 	}
 
 	// default read all data when rang is nil
-	var limitReadN int64 = hdfsFile.Stat().Size()
-
-	if rang != nil {
-		_, err = hdfsFile.Seek(int64(rang.StartIndex), 0)
+	var limitReadN = hdfsFile.Stat().Size()
+	if limitReadN < 0 {
+		return nil, errors.Errorf("file length is illegal, length: %d", limitReadN)
+	}
+	if request.Header.Get(source.Range) != "" {
+		requestRange, err := rangeutils.ParseRange(request.Header.Get(source.Range), uint64(limitReadN))
+		if err != nil {
+			return nil, err
+		}
+		_, err = hdfsFile.Seek(int64(requestRange.StartIndex), 0)
 		if err != nil {
 			hdfsFile.Close()
 			return nil, err
 		}
-		limitReadN = int64(rang.EndIndex - rang.StartIndex)
+		limitReadN = int64(requestRange.EndIndex - requestRange.StartIndex)
 	}
 
 	return newHdfsFileReaderClose(hdfsFile, limitReadN, hdfsFile), nil
 }
 
-func (h *hdfsSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
+func (h *hdfsSourceClient) DownloadWithExpireInfo(request *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
 
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -168,25 +160,31 @@ func (h *hdfsSourceClient) DownloadWithResponseHeader(ctx context.Context, url s
 	fileInfo := hdfsFile.Stat()
 
 	// default read all data when rang is nil
-	var limitReadN int64 = fileInfo.Size()
+	var limitReadN = fileInfo.Size()
+	if limitReadN < 0 {
+		return nil, nil, errors.Errorf("file length is illegal, length: %d", limitReadN)
+	}
 
-	if rang != nil {
-		_, err = hdfsFile.Seek(int64(rang.StartIndex), 0)
+	if request.Header.Get(source.Range) != "" {
+		requestRange, err := rangeutils.ParseRange(request.Header.Get(source.Range), uint64(limitReadN))
+		if err != nil {
+			return nil, nil, err
+		}
+		_, err = hdfsFile.Seek(int64(requestRange.StartIndex), 0)
 		if err != nil {
 			hdfsFile.Close()
 			return nil, nil, err
 		}
-		limitReadN = int64(rang.EndIndex - rang.StartIndex)
+		limitReadN = int64(requestRange.EndIndex - requestRange.StartIndex)
 	}
-
-	return newHdfsFileReaderClose(hdfsFile, limitReadN, hdfsFile), source.ResponseHeader{
-		source.LastModified: fileInfo.ModTime().Format(layout),
+	return newHdfsFileReaderClose(hdfsFile, limitReadN, hdfsFile), &source.ExpireInfo{
+		LastModified: timeutils.Format(fileInfo.ModTime()),
 	}, nil
 }
 
-func (h *hdfsSourceClient) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+func (h *hdfsSourceClient) GetLastModified(request *source.Request) (int64, error) {
 
-	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	hdfsClient, path, err := h.getHDFSClientAndPath(request.URL)
 	if err != nil {
 		return -1, err
 	}
@@ -200,19 +198,10 @@ func (h *hdfsSourceClient) GetLastModifiedMillis(ctx context.Context, url string
 }
 
 // getHDFSClient return hdfs client
-func (h *hdfsSourceClient) getHDFSClient(rawurl string) (*hdfs.Client, error) {
-	if len(rawurl) < 4 {
-		return nil, errors.Errorf("hdfs url invalid: url is %s", rawurl)
-	}
-
-	parse, err := url.Parse(rawurl)
-	if err != nil {
-		return nil, err
-	}
-
+func (h *hdfsSourceClient) getHDFSClient(url *url.URL) (*hdfs.Client, error) {
 	// get client for map
 	h.RWMutex.RLock()
-	if client, ok := h.clientMap[parse.Host]; ok {
+	if client, ok := h.clientMap[url.Host]; ok {
 		h.RWMutex.RUnlock()
 		return client, nil
 	}
@@ -222,7 +211,7 @@ func (h *hdfsSourceClient) getHDFSClient(rawurl string) (*hdfs.Client, error) {
 	options := hdfs.ClientOptionsFromConf(map[string]string{
 		hdfsUseDataNodeHostName: hdfsUseDataNodeHostNameValue,
 	})
-	options.Addresses = strings.Split(parse.Host, ",")
+	options.Addresses = strings.Split(url.Host, ",")
 	u, err := user.Current()
 	if err != nil {
 		return nil, err
@@ -236,34 +225,25 @@ func (h *hdfsSourceClient) getHDFSClient(rawurl string) (*hdfs.Client, error) {
 		h.RWMutex.Unlock()
 		return nil, err
 	}
-	h.clientMap[parse.Host] = client
+	h.clientMap[url.Host] = client
 	h.RWMutex.Unlock()
 	return client, err
 }
 
-// getHDFSPath return file path
-func (h *hdfsSourceClient) getHDFSPath(urls string) (string, error) {
-	parse, err := url.Parse(urls)
-	if err != nil {
-		return "", err
-	}
-	return parse.Path, nil
-}
-
 // getHDFSClientAndPath return client and path
-func (h *hdfsSourceClient) getHDFSClientAndPath(urls string) (*hdfs.Client, string, error) {
-	client, err := h.getHDFSClient(urls)
-	if err != nil {
-		return nil, "", errors.Errorf("hdfs create client fail, url is %s", urls)
-	}
-	path, err := h.getHDFSPath(urls)
+func (h *hdfsSourceClient) getHDFSClientAndPath(url *url.URL) (*hdfs.Client, string, error) {
+	client, err := h.getHDFSClient(url)
 	if err != nil {
-		return client, "", errors.Errorf("hdfs url path parse fail, url is %s", urls)
+		return nil, "", errors.Errorf("hdfs create client failed, url is %s", url)
 	}
-	return client, path, nil
+	return client, url.Path, nil
 }
 
 func NewHDFSSourceClient(opts ...HDFSSourceClientOption) source.ResourceClient {
+	return newHDFSSourceClient(opts...)
+}
+
+func newHDFSSourceClient(opts ...HDFSSourceClientOption) *hdfsSourceClient {
 	sourceClient := &hdfsSourceClient{
 		clientMap: make(map[string]*hdfs.Client),
 	}
diff --git a/pkg/source/header.go b/pkg/source/header.go
new file mode 100644
index 00000000..583152f4
--- /dev/null
+++ b/pkg/source/header.go
@@ -0,0 +1,120 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import "net/textproto"
+
+const (
+	LastModified    = "X-Dragonfly-Last-Modified"
+	IfModifiedSince = "X-Dragonfly-If-Modified-Since"
+	ETag            = "X-Dragonfly-ETag"
+	IfNoneMatch     = "X-Dragonfly-If-None-Match"
+	Range           = "X-Dragonfly-Range" // startIndex-endIndex
+)
+
+const LastModifiedLayout = "Mon, 02 Jan 2006 15:04:05 GMT"
+
+type ExpireInfo struct {
+	LastModified string // Mon, 02 Jan 2006 15:04:05 GMT
+	ETag         string
+}
+
+// A Header represents the key-value pairs in a Dragonfly source header.
+//
+// The keys should be in canonical form, as returned by
+// CanonicalHeaderKey.
+type Header map[string][]string
+
+// Add adds the key, value pair to the header.
+// It appends to any existing values associated with key.
+// The key is case insensitive; it is canonicalized by
+// CanonicalHeaderKey.
+func (h Header) Add(key, value string) {
+	textproto.MIMEHeader(h).Add(key, value)
+}
+
+// Set sets the header entries associated with key to the
+// single element value. It replaces any existing values
+// associated with key. The key is case insensitive; it is
+// canonicalized by textproto.CanonicalMIMEHeaderKey.
+// To use non-canonical keys, assign to the map directly.
+func (h Header) Set(key, value string) {
+	textproto.MIMEHeader(h).Set(key, value)
+}
+
+// Get gets the first value associated with the given key. If
+// there are no values associated with the key, Get returns "".
+// It is case insensitive; textproto.CanonicalMIMEHeaderKey is
+// used to canonicalize the provided key. To use non-canonical keys,
+// access the map directly.
+func (h Header) Get(key string) string {
+	return textproto.MIMEHeader(h).Get(key)
+}
+
+// Values returns all values associated with the given key.
+// It is case insensitive; textproto.CanonicalMIMEHeaderKey is
+// used to canonicalize the provided key. To use non-canonical
+// keys, access the map directly.
+// The returned slice is not a copy.
+func (h Header) Values(key string) []string {
+	return textproto.MIMEHeader(h).Values(key)
+}
+
+// get is like Get, but key must already be in CanonicalHeaderKey form.
+func (h Header) get(key string) string {
+	if v := h[key]; len(v) > 0 {
+		return v[0]
+	}
+	return ""
+}
+
+// has reports whether h has the provided key defined, even if it's
+// set to 0-length slice.
+func (h Header) has(key string) bool {
+	_, ok := h[key]
+	return ok
+}
+
+// Del deletes the values associated with key.
+// The key is case insensitive; it is canonicalized by
+// CanonicalHeaderKey.
+func (h Header) Del(key string) {
+	textproto.MIMEHeader(h).Del(key)
+}
+
+// Clone returns a copy of h or nil if h is nil.
+func (h Header) Clone() Header {
+	if h == nil {
+		return nil
+	}
+
+	// Find total number of values.
+	nv := 0
+	for _, vv := range h {
+		nv += len(vv)
+	}
+	sv := make([]string, nv) // shared backing array for headers' values
+	h2 := make(Header, len(h))
+	for k, vv := range h {
+		n := copy(sv, vv)
+		h2[k] = sv[:n:n]
+		sv = sv[n:]
+	}
+	return h2
+}
+
+func CanonicalHeaderKey(s string) string { return textproto.CanonicalMIMEHeaderKey(s) }
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/httpprotocol/http_source_client.go
index 6325558c..0b6d477c 100644
--- a/pkg/source/httpprotocol/http_source_client.go
+++ b/pkg/source/httpprotocol/http_source_client.go
@@ -17,7 +17,6 @@
 package httpprotocol
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"net"
@@ -28,11 +27,7 @@ import (
 
 	"github.com/go-http-utils/headers"
 
-	"d7y.io/dragonfly/v2/cdn/types"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/util/maputils"
-	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
-	"d7y.io/dragonfly/v2/pkg/util/stringutils"
 	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 )
 
@@ -74,8 +69,31 @@ func init() {
 		Transport: transport,
 	}
 	sc := NewHTTPSourceClient()
-	source.Register(HTTPClient, sc)
-	source.Register(HTTPSClient, sc)
+
+	if err := source.Register(HTTPClient, sc, Adapter); err != nil {
+		panic(err)
+	}
+
+	if err := source.Register(HTTPSClient, sc, Adapter); err != nil {
+		panic(err)
+	}
+}
+
+func Adapter(request *source.Request) *source.Request {
+	clonedRequest := request.Clone(request.Context())
+	if request.Header.Get(source.Range) != "" {
+		clonedRequest.Header.Set(headers.Range, fmt.Sprintf("bytes=%s", request.Header.Get(source.Range)))
+		clonedRequest.Header.Del(source.Range)
+	}
+	if request.Header.Get(source.LastModified) != "" {
+		clonedRequest.Header.Set(headers.LastModified, request.Header.Get(source.LastModified))
+		clonedRequest.Header.Del(source.LastModified)
+	}
+	if request.Header.Get(source.ETag) != "" {
+		clonedRequest.Header.Set(headers.ETag, request.Header.Get(source.ETag))
+		clonedRequest.Header.Del(source.ETag)
+	}
+	return clonedRequest
 }
 
 // httpSourceClient is an implementation of the interface of source.ResourceClient.
@@ -85,6 +103,10 @@ type httpSourceClient struct {
 
 // NewHTTPSourceClient returns a new HTTPSourceClientOption.
 func NewHTTPSourceClient(opts ...HTTPSourceClientOption) source.ResourceClient {
+	return newHTTPSourceClient(opts...)
+}
+
+func newHTTPSourceClient(opts ...HTTPSourceClientOption) *httpSourceClient {
 	client := &httpSourceClient{
 		httpClient: _defaultHTTPClient,
 	}
@@ -102,109 +124,107 @@ func WithHTTPClient(client *http.Client) HTTPSourceClientOption {
 	}
 }
 
-func (client *httpSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
-	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
+func (client *httpSourceClient) GetContentLength(request *source.Request) (int64, error) {
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
-		return -1, err
+		return source.UnKnownSourceFileLen, err
 	}
-	resp.Body.Close()
-	// TODO Here if other status codes should be added to ErrURLNotReachable, if not, it will be downloaded frequently for 404 or 403
-	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
-		// TODO Whether this situation should be distinguished from the err situation,
-		//similar to proposing another error type to indicate that this  error can interact with the URL, but the status code does not meet expectations
-		return types.IllegalSourceFileLen, fmt.Errorf("get http resource length failed, unexpected code: %d", resp.StatusCode)
+	defer resp.Body.Close()
+	err = source.CheckResponseCode(resp.StatusCode, []int{http.StatusOK, http.StatusPartialContent})
+	if err != nil {
+		return source.UnKnownSourceFileLen, err
 	}
 	return resp.ContentLength, nil
 }
 
-func (client *httpSourceClient) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
-	copied := maputils.DeepCopy(header)
-	copied[headers.Range] = "bytes=0-0"
-
-	resp, err := client.doRequest(ctx, http.MethodGet, url, copied)
+func (client *httpSourceClient) IsSupportRange(request *source.Request) (bool, error) {
+	if request.Header.Get(headers.Range) == "" {
+		request.Header.Set(headers.Range, "bytes=0-0")
+	}
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
 		return false, err
 	}
-	resp.Body.Close()
+	defer resp.Body.Close()
 	return resp.StatusCode == http.StatusPartialContent, nil
 }
 
-// TODO Consider the situation where there is no last-modified such as baidu
-func (client *httpSourceClient) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
-	lastModified := timeutils.UnixMillis(expireInfo[source.LastModified])
-
-	eTag := expireInfo[headers.ETag]
-	if lastModified <= 0 && stringutils.IsBlank(eTag) {
-		return true, nil
-	}
-
-	// set header: header is a reference to map, should not change it
-	copied := maputils.DeepCopy(header)
-	if lastModified > 0 {
-		copied[headers.IfModifiedSince] = expireInfo[headers.LastModified]
-	}
-	if !stringutils.IsBlank(eTag) {
-		copied[headers.IfNoneMatch] = eTag
+func (client *httpSourceClient) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
+	if info != nil {
+		if request.Header == nil {
+			request.Header = source.Header{}
+		}
+		if info.LastModified != "" {
+			request.Header.Set(headers.IfModifiedSince, info.LastModified)
+		}
+		if info.ETag != "" {
+			request.Header.Set(headers.IfNoneMatch, info.ETag)
+		}
 	}
-
-	// send request
-	resp, err := client.doRequest(ctx, http.MethodGet, url, copied)
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
-		// If it fails to get the result, it is considered that the source has not expired, to prevent the source from exploding
 		return false, err
 	}
-	resp.Body.Close()
-	return resp.StatusCode != http.StatusNotModified, nil
+	defer resp.Body.Close()
+	return !(resp.StatusCode == http.StatusNotModified || (resp.Header.Get(headers.ETag) == info.ETag || resp.Header.Get(headers.LastModified) == info.
+		LastModified)), nil
 }
 
-func (client *httpSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
-	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
+func (client *httpSourceClient) Download(request *source.Request) (io.ReadCloser, error) {
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
 		return nil, err
 	}
-	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusPartialContent {
-		return resp.Body, nil
+	err = source.CheckResponseCode(resp.StatusCode, []int{http.StatusOK, http.StatusPartialContent})
+	if err != nil {
+		resp.Body.Close()
+		return nil, err
 	}
-	resp.Body.Close()
-	return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	return resp.Body, nil
 }
 
-func (client *httpSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader,
-	error) {
-	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
+func (client *httpSourceClient) DownloadWithExpireInfo(request *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
 		return nil, nil, err
 	}
-	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusPartialContent {
-		responseHeader := source.ResponseHeader{
-			source.LastModified: resp.Header.Get(headers.LastModified),
-			source.ETag:         resp.Header.Get(headers.ETag),
-		}
-		return resp.Body, responseHeader, nil
+	err = source.CheckResponseCode(resp.StatusCode, []int{http.StatusOK, http.StatusPartialContent})
+	if err != nil {
+		resp.Body.Close()
+		return nil, nil, err
 	}
-	resp.Body.Close()
-	return nil, nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	return resp.Body, &source.ExpireInfo{
+		LastModified: resp.Header.Get(headers.LastModified),
+		ETag:         resp.Header.Get(headers.ETag),
+	}, nil
 }
 
-func (client *httpSourceClient) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
-	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
+func (client *httpSourceClient) GetLastModified(request *source.Request) (int64, error) {
+	resp, err := client.doRequest(http.MethodGet, request)
 	if err != nil {
 		return -1, err
 	}
-	resp.Body.Close()
-	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusPartialContent {
-		return timeutils.UnixMillis(resp.Header.Get(headers.LastModified)), nil
+	defer resp.Body.Close()
+	err = source.CheckResponseCode(resp.StatusCode, []int{http.StatusOK, http.StatusPartialContent})
+	if err != nil {
+		return -1, err
 	}
-	return -1, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	return timeutils.UnixMillis(resp.Header.Get(headers.LastModified)), nil
 }
 
-func (client *httpSourceClient) doRequest(ctx context.Context, method string, url string, header source.RequestHeader) (*http.Response, error) {
-	req, err := http.NewRequestWithContext(ctx, method, url, nil)
+func (client *httpSourceClient) doRequest(method string, request *source.Request) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(request.Context(), method, request.URL.String(), nil)
 	if err != nil {
 		return nil, err
 	}
-	for k, v := range header {
-		req.Header.Add(k, v)
+	for key, values := range request.Header {
+		for i := range values {
+			req.Header.Add(key, values[i])
+		}
+	}
+	resp, err := client.httpClient.Do(req)
+	if err != nil {
+		return nil, err
 	}
-	return client.httpClient.Do(req)
+	return resp, nil
 }
diff --git a/pkg/source/mock/mock_source_client.go b/pkg/source/mock/mock_source_client.go
index 0912f60e..bd4d0293 100644
--- a/pkg/source/mock/mock_source_client.go
+++ b/pkg/source/mock/mock_source_client.go
@@ -5,12 +5,10 @@
 package mock
 
 import (
-	context "context"
 	io "io"
 	reflect "reflect"
 
 	source "d7y.io/dragonfly/v2/pkg/source"
-	rangeutils "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 	gomock "github.com/golang/mock/gomock"
 )
 
@@ -38,92 +36,92 @@ func (m *MockResourceClient) EXPECT() *MockResourceClientMockRecorder {
 }
 
 // Download mocks base method.
-func (m *MockResourceClient) Download(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (io.ReadCloser, error) {
+func (m *MockResourceClient) Download(arg0 *source.Request) (io.ReadCloser, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Download", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "Download", arg0)
 	ret0, _ := ret[0].(io.ReadCloser)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // Download indicates an expected call of Download.
-func (mr *MockResourceClientMockRecorder) Download(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) Download(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Download", reflect.TypeOf((*MockResourceClient)(nil).Download), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Download", reflect.TypeOf((*MockResourceClient)(nil).Download), arg0)
 }
 
-// DownloadWithResponseHeader mocks base method.
-func (m *MockResourceClient) DownloadWithResponseHeader(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
+// DownloadWithExpireInfo mocks base method.
+func (m *MockResourceClient) DownloadWithExpireInfo(arg0 *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DownloadWithResponseHeader", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "DownloadWithExpireInfo", arg0)
 	ret0, _ := ret[0].(io.ReadCloser)
-	ret1, _ := ret[1].(source.ResponseHeader)
+	ret1, _ := ret[1].(*source.ExpireInfo)
 	ret2, _ := ret[2].(error)
 	return ret0, ret1, ret2
 }
 
-// DownloadWithResponseHeader indicates an expected call of DownloadWithResponseHeader.
-func (mr *MockResourceClientMockRecorder) DownloadWithResponseHeader(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+// DownloadWithExpireInfo indicates an expected call of DownloadWithExpireInfo.
+func (mr *MockResourceClientMockRecorder) DownloadWithExpireInfo(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadWithResponseHeader", reflect.TypeOf((*MockResourceClient)(nil).DownloadWithResponseHeader), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadWithExpireInfo", reflect.TypeOf((*MockResourceClient)(nil).DownloadWithExpireInfo), arg0)
 }
 
 // GetContentLength mocks base method.
-func (m *MockResourceClient) GetContentLength(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (int64, error) {
+func (m *MockResourceClient) GetContentLength(arg0 *source.Request) (int64, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetContentLength", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "GetContentLength", arg0)
 	ret0, _ := ret[0].(int64)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // GetContentLength indicates an expected call of GetContentLength.
-func (mr *MockResourceClientMockRecorder) GetContentLength(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) GetContentLength(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockResourceClient)(nil).GetContentLength), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockResourceClient)(nil).GetContentLength), arg0)
 }
 
-// GetLastModifiedMillis mocks base method.
-func (m *MockResourceClient) GetLastModifiedMillis(arg0 context.Context, arg1 string, arg2 source.RequestHeader) (int64, error) {
+// GetLastModified mocks base method.
+func (m *MockResourceClient) GetLastModified(arg0 *source.Request) (int64, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetLastModifiedMillis", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "GetLastModified", arg0)
 	ret0, _ := ret[0].(int64)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
-// GetLastModifiedMillis indicates an expected call of GetLastModifiedMillis.
-func (mr *MockResourceClientMockRecorder) GetLastModifiedMillis(arg0, arg1, arg2 interface{}) *gomock.Call {
+// GetLastModified indicates an expected call of GetLastModified.
+func (mr *MockResourceClientMockRecorder) GetLastModified(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetLastModifiedMillis", reflect.TypeOf((*MockResourceClient)(nil).GetLastModifiedMillis), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetLastModified", reflect.TypeOf((*MockResourceClient)(nil).GetLastModified), arg0)
 }
 
 // IsExpired mocks base method.
-func (m *MockResourceClient) IsExpired(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 map[string]string) (bool, error) {
+func (m *MockResourceClient) IsExpired(arg0 *source.Request, arg1 *source.ExpireInfo) (bool, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "IsExpired", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "IsExpired", arg0, arg1)
 	ret0, _ := ret[0].(bool)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // IsExpired indicates an expected call of IsExpired.
-func (mr *MockResourceClientMockRecorder) IsExpired(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) IsExpired(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsExpired", reflect.TypeOf((*MockResourceClient)(nil).IsExpired), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsExpired", reflect.TypeOf((*MockResourceClient)(nil).IsExpired), arg0, arg1)
 }
 
 // IsSupportRange mocks base method.
-func (m *MockResourceClient) IsSupportRange(arg0 context.Context, arg1 string, arg2 source.RequestHeader) (bool, error) {
+func (m *MockResourceClient) IsSupportRange(arg0 *source.Request) (bool, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "IsSupportRange", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "IsSupportRange", arg0)
 	ret0, _ := ret[0].(bool)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // IsSupportRange indicates an expected call of IsSupportRange.
-func (mr *MockResourceClientMockRecorder) IsSupportRange(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) IsSupportRange(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsSupportRange", reflect.TypeOf((*MockResourceClient)(nil).IsSupportRange), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsSupportRange", reflect.TypeOf((*MockResourceClient)(nil).IsSupportRange), arg0)
 }
diff --git a/pkg/source/ossprotocol/oss_source_client.go b/pkg/source/ossprotocol/oss_source_client.go
index 0671f253..8a172521 100644
--- a/pkg/source/ossprotocol/oss_source_client.go
+++ b/pkg/source/ossprotocol/oss_source_client.go
@@ -17,11 +17,9 @@
 package ossprotocol
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"net/http"
-	"net/url"
 	"strconv"
 	"sync"
 
@@ -29,13 +27,12 @@ import (
 	"github.com/go-http-utils/headers"
 	"github.com/pkg/errors"
 
-	cdnerrors "d7y.io/dragonfly/v2/cdn/errors"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
+	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 )
 
-const ossClient = "oss"
+const OSSClient = "oss"
 
 const (
 	endpoint        = "endpoint"
@@ -46,11 +43,33 @@ const (
 var _ source.ResourceClient = (*ossSourceClient)(nil)
 
 func init() {
-	sourceClient := NewOSSSourceClient()
-	source.Register(ossClient, sourceClient)
+	if err := source.Register(OSSClient, NewOSSSourceClient(), adaptor); err != nil {
+		panic(err)
+	}
 }
 
-func NewOSSSourceClient(opts ...OssSourceClientOption) source.ResourceClient {
+func adaptor(request *source.Request) *source.Request {
+	clonedRequest := request.Clone(request.Context())
+	if request.Header.Get(source.Range) != "" {
+		clonedRequest.Header.Set(oss.HTTPHeaderRange, fmt.Sprintf("bytes=%s", request.Header.Get(source.Range)))
+		clonedRequest.Header.Del(source.Range)
+	}
+	if request.Header.Get(source.LastModified) != "" {
+		clonedRequest.Header.Set(oss.HTTPHeaderLastModified, request.Header.Get(source.LastModified))
+		clonedRequest.Header.Del(source.LastModified)
+	}
+	if request.Header.Get(source.ETag) != "" {
+		clonedRequest.Header.Set(oss.HTTPHeaderEtag, request.Header.Get(source.ETag))
+		clonedRequest.Header.Del(source.ETag)
+	}
+	return clonedRequest
+}
+
+func NewOSSSourceClient(opts ...OSSSourceClientOption) source.ResourceClient {
+	return newOSSSourceClient(opts...)
+}
+
+func newOSSSourceClient(opts ...OSSSourceClientOption) source.ResourceClient {
 	sourceClient := &ossSourceClient{
 		clientMap: sync.Map{},
 		accessMap: sync.Map{},
@@ -61,103 +80,143 @@ func NewOSSSourceClient(opts ...OssSourceClientOption) source.ResourceClient {
 	return sourceClient
 }
 
-type OssSourceClientOption func(p *ossSourceClient)
+type OSSSourceClientOption func(p *ossSourceClient)
 
-// ossSourceClient is an implementation of the interface of SourceClient.
+// ossSourceClient is an implementation of the interface of source.ResourceClient.
 type ossSourceClient struct {
 	// endpoint_accessKeyID_accessKeySecret -> ossClient
 	clientMap sync.Map
 	accessMap sync.Map
 }
 
-func (osc *ossSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
-	panic("implement me")
-}
-
-func (osc *ossSourceClient) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
-	panic("implement me")
-}
-
-func (osc *ossSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
-	resHeader, err := osc.getMeta(ctx, url, header)
+func (osc *ossSourceClient) GetContentLength(request *source.Request) (int64, error) {
+	client, err := osc.getClient(request.Header)
 	if err != nil {
-		return -1, err
+		return source.UnKnownSourceFileLen, err
 	}
-
-	contentLen, err := strconv.ParseInt(resHeader.Get(oss.HTTPHeaderContentLength), 10, 64)
+	bucket, err := client.Bucket(request.URL.Host)
 	if err != nil {
-		return -1, err
+		return source.UnKnownSourceFileLen, errors.Wrapf(err, "get oss bucket: %s", request.URL.Host)
+	}
+	header, err := bucket.GetObjectMeta(request.URL.Path, getOptions(request.Header)...)
+	if err != nil {
+		return source.UnKnownSourceFileLen, errors.Wrapf(err, "get oss object %s meta", request.URL.Path)
+	}
+	contentLen, err := strconv.ParseInt(header.Get(oss.HTTPHeaderContentLength), 10, 64)
+	if err != nil {
+		return source.UnKnownSourceFileLen, errors.Wrapf(err, "parse content-length str to int64")
 	}
-
 	return contentLen, nil
 }
 
-func (osc *ossSourceClient) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
-	_, err := osc.getMeta(ctx, url, header)
+func (osc *ossSourceClient) IsSupportRange(request *source.Request) (bool, error) {
+	if request.Header.Get(oss.HTTPHeaderRange) == "" {
+		request.Header.Set(oss.HTTPHeaderRange, "bytes=0-0")
+	}
+	client, err := osc.getClient(request.Header)
+	if err != nil {
+		return false, errors.Wrap(err, "get oss client")
+	}
+	bucket, err := client.Bucket(request.URL.Host)
+	if err != nil {
+		return false, errors.Wrapf(err, "get oss bucket: %s", request.URL.Host)
+	}
+	exist, err := bucket.IsObjectExist(request.URL.Path, getOptions(request.Header)...)
 	if err != nil {
 		return false, err
 	}
+	if !exist {
+		return false, source.ErrResourceNotReachable
+	}
 	return true, nil
 }
 
-func (osc *ossSourceClient) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
-	lastModified := expireInfo[oss.HTTPHeaderLastModified]
-	eTag := expireInfo[oss.HTTPHeaderEtag]
-	if stringutils.IsBlank(lastModified) && stringutils.IsBlank(eTag) {
-		return true, nil
+func (osc *ossSourceClient) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
+	client, err := osc.getClient(request.Header)
+	if err != nil {
+		return false, errors.Wrap(err, "get oss client")
 	}
-
-	resHeader, err := osc.getMeta(ctx, url, header)
+	bucket, err := client.Bucket(request.URL.Host)
+	if err != nil {
+		return false, errors.Wrapf(err, "get oss bucket: %s", request.URL.Host)
+	}
+	resHeader, err := bucket.GetObjectMeta(request.URL.Path, getOptions(request.Header)...)
 	if err != nil {
 		return false, err
 	}
-	return resHeader.Get(oss.HTTPHeaderLastModified) == expireInfo[oss.HTTPHeaderLastModified] && resHeader.Get(oss.HTTPHeaderEtag) == expireInfo[oss.
-		HTTPHeaderEtag], nil
+	return !(resHeader.Get(oss.HTTPHeaderEtag) == info.ETag || resHeader.Get(oss.HTTPHeaderLastModified) == info.LastModified), nil
 }
 
-func (osc *ossSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
-	ossObject, err := parseOssObject(url)
+func (osc *ossSourceClient) Download(request *source.Request) (io.ReadCloser, error) {
+	client, err := osc.getClient(request.Header)
 	if err != nil {
-		return nil, nil, errors.Wrapf(err, "parse oss object from url: %s", url)
+		return nil, errors.Wrap(err, "get oss client")
 	}
-	client, err := osc.getClient(header)
+	bucket, err := client.Bucket(request.URL.Host)
 	if err != nil {
-		return nil, nil, errors.Wrapf(err, "failed to get client")
+		return nil, errors.Wrapf(err, "get oss bucket %s", request.URL.Host)
 	}
-	bucket, err := client.Bucket(ossObject.bucket)
+	resp, err := bucket.GetObject(request.URL.Path, getOptions(request.Header)...)
 	if err != nil {
-		return nil, nil, errors.Wrapf(err, "failed to get bucket: %s", ossObject.bucket)
+		return nil, errors.Wrapf(err, "get oss object %s", request.URL.Path)
 	}
-	res, err := bucket.GetObject(ossObject.object, getOptions(header)...)
+	return resp, nil
+}
+
+func (osc *ossSourceClient) DownloadWithExpireInfo(request *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
+	client, err := osc.getClient(request.Header)
 	if err != nil {
-		return nil, nil, errors.Wrapf(err, "failed to get oss Object: %s", ossObject.object)
+		return nil, nil, errors.Wrapf(err, "get oss client")
 	}
-	resp := res.(*oss.Response)
-	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusPartialContent {
-		responseHeader := source.ResponseHeader{
-			source.LastModified: resp.Headers.Get(headers.LastModified),
-			source.ETag:         resp.Headers.Get(headers.ETag),
-		}
-		return resp.Body, responseHeader, nil
+	bucket, err := client.Bucket(request.URL.Host)
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, "get oss bucket: %s", request.URL.Host)
 	}
-	resp.Body.Close()
-	return nil, nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	objectResult, err := bucket.DoGetObject(&oss.GetObjectRequest{ObjectKey: request.URL.Path}, getOptions(request.Header))
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, "get oss Object: %s", request.URL.Path)
+	}
+	err = source.CheckResponseCode(objectResult.Response.StatusCode, []int{http.StatusOK, http.StatusPartialContent})
+	if err != nil {
+		objectResult.Response.Body.Close()
+		return nil, nil, err
+	}
+	return objectResult.Response.Body, &source.ExpireInfo{
+		LastModified: objectResult.Response.Headers.Get(headers.LastModified),
+		ETag:         objectResult.Response.Headers.Get(headers.ETag),
+	}, nil
 }
 
-func (osc *ossSourceClient) getClient(header map[string]string) (*oss.Client, error) {
-	endpoint, ok := header[endpoint]
-	if !ok {
-		return nil, errors.Wrapf(cdnerrors.ErrInvalidValue, "endpoint is empty")
+func (osc *ossSourceClient) GetLastModified(request *source.Request) (int64, error) {
+	client, err := osc.getClient(request.Header)
+	if err != nil {
+		return -1, errors.Wrap(err, "get oss client")
 	}
-	accessKeyID, ok := header[accessKeyID]
-	if !ok {
-		return nil, errors.Wrapf(cdnerrors.ErrInvalidValue, "accessKeyID is empty")
+	bucket, err := client.Bucket(request.URL.Host)
+	if err != nil {
+		return -1, errors.Wrapf(err, "get oss bucket: %s", request.URL.Host)
 	}
-	accessKeySecret, ok := header[accessKeySecret]
-	if !ok {
-		return nil, errors.Wrapf(cdnerrors.ErrInvalidValue, "accessKeySecret is empty")
+	respHeader, err := bucket.GetObjectMeta(request.URL.Path, getOptions(request.Header)...)
+	if err != nil {
+		return -1, err
 	}
-	clientKey := genClientKey(endpoint, accessKeyID, accessKeySecret)
+	return timeutils.UnixMillis(respHeader.Get(oss.HTTPHeaderLastModified)), nil
+}
+
+func (osc *ossSourceClient) getClient(header source.Header) (*oss.Client, error) {
+	endpoint := header.Get(endpoint)
+	if stringutils.IsBlank(endpoint) {
+		return nil, errors.New("endpoint is empty")
+	}
+	accessKeyID := header.Get(accessKeyID)
+	if stringutils.IsBlank(accessKeyID) {
+		return nil, errors.New("accessKeyID is empty")
+	}
+	accessKeySecret := header.Get(accessKeySecret)
+	if stringutils.IsBlank(accessKeySecret) {
+		return nil, errors.New("accessKeySecret is empty")
+	}
+	clientKey := buildClientKey(endpoint, accessKeyID, accessKeySecret)
 	if client, ok := osc.clientMap.Load(clientKey); ok {
 		return client.(*oss.Client), nil
 	}
@@ -165,39 +224,15 @@ func (osc *ossSourceClient) getClient(header map[string]string) (*oss.Client, er
 	if err != nil {
 		return nil, err
 	}
-	osc.clientMap.Store(clientKey, client)
-	return client, nil
+	actual, _ := osc.clientMap.LoadOrStore(clientKey, client)
+	return actual.(*oss.Client), nil
 }
 
-func genClientKey(endpoint, accessKeyID, accessKeySecret string) string {
+func buildClientKey(endpoint, accessKeyID, accessKeySecret string) string {
 	return fmt.Sprintf("%s_%s_%s", endpoint, accessKeyID, accessKeySecret)
 }
 
-func (osc *ossSourceClient) getMeta(ctx context.Context, url string, header map[string]string) (http.Header, error) {
-	client, err := osc.getClient(header)
-	if err != nil {
-		return nil, errors.Wrapf(err, "get oss client")
-	}
-	ossObject, err := parseOssObject(url)
-	if err != nil {
-		return nil, errors.Wrapf(err, "parse oss object")
-	}
-
-	bucket, err := client.Bucket(ossObject.bucket)
-	if err != nil {
-		return nil, errors.Wrapf(err, "get bucket: %s", ossObject.bucket)
-	}
-	isExist, err := bucket.IsObjectExist(ossObject.object)
-	if err != nil {
-		return nil, errors.Wrapf(err, "prob object: %s if exist", ossObject.object)
-	}
-	if !isExist {
-		return nil, fmt.Errorf("oss object: %s does not exist", ossObject.object)
-	}
-	return bucket.GetObjectMeta(ossObject.object, getOptions(header)...)
-}
-
-func getOptions(header map[string]string) []oss.Option {
+func getOptions(header source.Header) []oss.Option {
 	opts := make([]oss.Option, 0, len(header))
 	for key, value := range header {
 		if key == endpoint || key == accessKeyID || key == accessKeySecret {
@@ -207,24 +242,3 @@ func getOptions(header map[string]string) []oss.Option {
 	}
 	return opts
 }
-
-type ossObject struct {
-	endpoint string
-	bucket   string
-	object   string
-}
-
-func parseOssObject(rawURL string) (*ossObject, error) {
-	parsedURL, err := url.Parse(rawURL)
-	if err != nil {
-		return nil, errors.Wrapf(err, "parse rawURL: %s failed", rawURL)
-	}
-	if parsedURL.Scheme != "oss" {
-		return nil, fmt.Errorf("rawUrl: %s is not oss url", rawURL)
-	}
-	return &ossObject{
-		endpoint: parsedURL.Path[0:2],
-		bucket:   parsedURL.Host,
-		object:   parsedURL.Path[1:],
-	}, nil
-}
diff --git a/pkg/source/plugin.go b/pkg/source/plugin.go
index 2a3c061f..72f88631 100644
--- a/pkg/source/plugin.go
+++ b/pkg/source/plugin.go
@@ -24,20 +24,20 @@ import (
 )
 
 const (
-	pluginMetadataSchema = "schema"
+	pluginMetadataScheme = "scheme"
 )
 
-func LoadPlugin(schema string) (ResourceClient, error) {
+func LoadPlugin(scheme string) (ResourceClient, error) {
 	// TODO init option
-	logger.Debugf("try to load source plugin: %s", schema)
-	client, meta, err := dfplugin.Load(dfplugin.PluginTypeResource, schema, map[string]string{})
+	logger.Debugf("try to load source plugin: %s", scheme)
+	client, meta, err := dfplugin.Load(dfplugin.PluginTypeResource, scheme, map[string]string{})
 	if err != nil {
 		logger.Errorf("load source plugin error: %s", err)
 		return nil, err
 	}
 
-	if meta[pluginMetadataSchema] != schema {
-		logger.Errorf("load source plugin error: support schema not match")
+	if meta[pluginMetadataScheme] != scheme {
+		logger.Errorf("load source plugin error: support scheme not match")
 		return nil, errors.New("support schema not match")
 	}
 
@@ -47,6 +47,6 @@ func LoadPlugin(schema string) (ResourceClient, error) {
 		return nil, errors.New("invalid client, not a ResourceClient")
 	}
 
-	logger.Debugf("loaded source plugin %s", schema)
+	logger.Debugf("loaded source plugin %s", scheme)
 	return rc, nil
 }
diff --git a/pkg/source/request.go b/pkg/source/request.go
new file mode 100644
index 00000000..edc38185
--- /dev/null
+++ b/pkg/source/request.go
@@ -0,0 +1,149 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"context"
+	"net/url"
+
+	"github.com/pkg/errors"
+)
+
+type Request struct {
+	URL    *url.URL
+	Header Header
+	// ctx is either the client or server context. It should only
+	// be modified via copying the whole Request using WithContext.
+	// It is unexported to prevent people from using Context wrong
+	// and mutating the contexts held by callers of the same request.
+	ctx context.Context
+}
+
+func NewRequest(rawURL string) (*Request, error) {
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		return nil, err
+	}
+	return &Request{
+		URL:    u,
+		Header: make(Header),
+		ctx:    context.Background(),
+	}, nil
+}
+
+func NewRequestWithHeader(rawURL string, header map[string]string) (*Request, error) {
+	request, err := NewRequest(rawURL)
+	if err != nil {
+		return nil, err
+	}
+	for k, v := range header {
+		request.Header.Add(k, v)
+	}
+	return request, nil
+}
+
+func NewRequestWithContext(ctx context.Context, rawURL string, header map[string]string) (*Request, error) {
+	if ctx == nil {
+		return nil, errors.New("nil Context")
+	}
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		return nil, err
+	}
+	req := &Request{
+		ctx:    ctx,
+		URL:    u,
+		Header: make(Header),
+	}
+	for k, v := range header {
+		req.Header.Add(k, v)
+	}
+	return req, nil
+}
+
+// Context returns the request's context. To change the context, use
+// WithContext.
+//
+// The returned context is always non-nil; it defaults to the
+// background context.
+//
+// For outgoing client requests, the context controls cancellation.
+//
+// For incoming server requests, the context is canceled when the
+// client's connection closes, the request is canceled (with HTTP/2),
+// or when the ServeHTTP method returns.
+func (r *Request) Context() context.Context {
+	if r.ctx != nil {
+		return r.ctx
+	}
+	return context.Background()
+}
+
+// WithContext returns a shallow copy of r with its context changed
+// to ctx. The provided ctx must be non-nil.
+//
+// For outgoing client request, the context controls the entire
+// lifetime of a request and its response: obtaining a connection,
+// sending the request, and reading the response headers and body.
+//
+// To create a new request with a context, use NewRequestWithContext.
+// To change the context of a request, such as an incoming request you
+// want to modify before sending back out, use Request.Clone. Between
+// those two uses, it's rare to need WithContext.
+func (r *Request) WithContext(ctx context.Context) *Request {
+	if ctx == nil {
+		panic("nil context")
+	}
+	r2 := new(Request)
+	*r2 = *r
+	r2.ctx = ctx
+	r2.URL = cloneURL(r.URL) // legacy behavior; TODO: try to remove. Issue 23544
+	return r2
+}
+
+func cloneURL(u *url.URL) *url.URL {
+	if u == nil {
+		return nil
+	}
+	u2 := new(url.URL)
+	*u2 = *u
+	if u.User != nil {
+		u2.User = new(url.Userinfo)
+		*u2.User = *u.User
+	}
+	return u2
+}
+
+// Clone returns a deep copy of r with its context changed to ctx.
+// The provided ctx must be non-nil.
+//
+// For an outgoing client request, the context controls the entire
+// lifetime of a request and its response: obtaining a connection,
+// sending the request, and reading the response headers and body.
+func (r *Request) Clone(ctx context.Context) *Request {
+	if ctx == nil {
+		panic("nil context")
+	}
+	r2 := new(Request)
+	*r2 = *r
+	r2.ctx = ctx
+	r2.URL = cloneURL(r.URL)
+	if r.Header != nil {
+		r2.Header = r.Header.Clone()
+	}
+	return r2
+}
diff --git a/pkg/source/request_fixture.go b/pkg/source/request_fixture.go
new file mode 100644
index 00000000..f179d1ad
--- /dev/null
+++ b/pkg/source/request_fixture.go
@@ -0,0 +1,42 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"fmt"
+
+	"github.com/golang/mock/gomock"
+)
+
+// RequestEq for gomock
+func RequestEq(url string) gomock.Matcher {
+	return requestMatcher{url}
+}
+
+type requestMatcher struct {
+	url string
+}
+
+// Matches returns whether x is a match.
+func (req requestMatcher) Matches(x interface{}) bool {
+	return x.(*Request).URL.String() == req.url
+}
+
+// String describes what the matcher matches.
+func (req requestMatcher) String() string {
+	return fmt.Sprintf("is equal to %v (%T)", req.url, req.url)
+}
diff --git a/pkg/source/headers.go b/pkg/source/response.go
similarity index 61%
rename from pkg/source/headers.go
rename to pkg/source/response.go
index 76ffdc96..cb1c86b1 100644
--- a/pkg/source/headers.go
+++ b/pkg/source/response.go
@@ -16,27 +16,12 @@
 
 package source
 
-// RequestHeader request header
-type RequestHeader map[string]string
+import "io"
 
-func (h RequestHeader) Get(key string) string {
-	if h == nil {
-		return ""
-	}
-	return h[key]
+type Response struct {
+	Status        string
+	StatusCode    int
+	Header        Header
+	Body          io.ReadCloser
+	ContentLength int64
 }
-
-// ResponseHeader response header
-type ResponseHeader map[string]string
-
-func (h ResponseHeader) Get(key string) string {
-	if h == nil {
-		return ""
-	}
-	return h[key]
-}
-
-const (
-	LastModified = "Last-Modified"
-	ETag         = "ETag"
-)
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index 5d82fb6a..03af6817 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -19,203 +19,271 @@ package source
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/url"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/go-http-utils/headers"
 	"github.com/pkg/errors"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
-var _ ResourceClient = (*clientManager)(nil)
-var _ ClientManager = (*clientManager)(nil)
+var (
+	// ErrResourceNotReachable represents the url resource is a not reachable.
+	ErrResourceNotReachable = errors.New("resource is not reachable")
+
+	// ErrNoClientFound represents no source client to resolve url
+	ErrNoClientFound = errors.New("no source client found")
+)
+
+// UnexpectedStatusCodeError is returned when a source responds with neither an error
+// nor with a status code indicating success.
+type UnexpectedStatusCodeError struct {
+	allowed []int // The expected stats code returned from source
+	got     int   // The actual status code from source
+}
+
+// Error implements interface error
+func (e UnexpectedStatusCodeError) Error() string {
+	var expected []string
+	for _, v := range e.allowed {
+		expected = append(expected, strconv.Itoa(v))
+	}
+	return fmt.Sprintf("status code from source is %s; was expecting %s",
+		strconv.Itoa(e.got), strings.Join(expected, " or "))
+}
+
+// Got is the actual status code returned by source.
+func (e UnexpectedStatusCodeError) Got() int {
+	return e.got
+}
 
-// ResourceClient supply apis that interact with the source.
+// CheckResponseCode returns UnexpectedStatusError if the given response code is not
+// one of the allowed status codes; otherwise nil.
+func CheckResponseCode(respCode int, allowed []int) error {
+	for _, v := range allowed {
+		if respCode == v {
+			return nil
+		}
+	}
+	return UnexpectedStatusCodeError{allowed, respCode}
+}
+
+func IsResourceNotReachableError(err error) bool {
+	return errors.Is(err, ErrResourceNotReachable)
+}
+
+func IsNoClientFoundError(err error) bool {
+	return errors.Is(err, ErrNoClientFound)
+}
+
+const (
+	UnKnownSourceFileLen = -2
+)
+
+// ResourceClient defines the API interface to interact with source.
 type ResourceClient interface {
 
 	// GetContentLength get length of resource content
-	// return -l if request fail
-	// return task.IllegalSourceFileLen if response status is not StatusOK and StatusPartialContent
-	GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error)
+	// return source.UnKnownSourceFileLen if response status is not StatusOK and StatusPartialContent
+	GetContentLength(request *Request) (int64, error)
 
 	// IsSupportRange checks if resource supports breakpoint continuation
-	IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error)
+	// return false if response status is not StatusPartialContent
+	IsSupportRange(request *Request) (bool, error)
 
 	// IsExpired checks if a resource received or stored is the same.
-	IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error)
+	// return false and non-nil err to prevent the source from exploding if
+	// fails to get the result, it is considered that the source has not expired
+	IsExpired(request *Request, info *ExpireInfo) (bool, error)
 
 	// Download downloads from source
-	Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error)
+	Download(request *Request) (io.ReadCloser, error)
 
-	// DownloadWithResponseHeader download from source with responseHeader
-	DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader, error)
+	// DownloadWithExpireInfo download from source with expireInfo
+	DownloadWithExpireInfo(request *Request) (io.ReadCloser, *ExpireInfo, error)
 
-	// GetLastModifiedMillis gets last modified timestamp milliseconds of resource
-	GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error)
+	// GetLastModified gets last modified timestamp milliseconds of resource
+	GetLastModified(request *Request) (int64, error)
 }
 
 type ClientManager interface {
-	ResourceClient
-	Register(schema string, resourceClient ResourceClient)
-	UnRegister(schema string)
+	// Register a source client with scheme
+	Register(scheme string, resourceClient ResourceClient, adapter requestAdapter, hook ...Hook) error
+
+	// UnRegister a source client from manager
+	UnRegister(scheme string)
+
+	// GetClient a source client by scheme
+	GetClient(scheme string) (ResourceClient, bool)
 }
 
+// clientManager implements the interface ClientManager
 type clientManager struct {
-	sync.RWMutex
+	mu      sync.RWMutex
 	clients map[string]ResourceClient
 }
 
+var _ ClientManager = (*clientManager)(nil)
+
 var _defaultManager = NewManager()
 
-func (m *clientManager) GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return -1, err
-	}
-	if _, ok := ctx.Deadline(); !ok {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
-		defer cancel()
+func NewManager() ClientManager {
+	return &clientManager{
+		clients: make(map[string]ResourceClient),
 	}
-	return sourceClient.GetContentLength(ctx, url, header, rang)
 }
 
-func (m *clientManager) IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return false, err
+func (m *clientManager) Register(scheme string, resourceClient ResourceClient, adaptor requestAdapter, hooks ...Hook) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if client, ok := m.clients[strings.ToLower(scheme)]; ok {
+		if client.(*clientWrapper).rc != resourceClient {
+			return errors.Errorf("client with scheme %s already exist, current client: %#v", scheme, client)
+		}
 	}
-	if _, ok := ctx.Deadline(); !ok {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
-		defer cancel()
+	m.clients[strings.ToLower(scheme)] = &clientWrapper{
+		adapter: adaptor,
+		hooks:   hooks,
+		rc:      resourceClient,
 	}
-	return sourceClient.IsSupportRange(ctx, url, header)
+	return nil
 }
 
-func (m *clientManager) IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return false, err
-	}
-	if _, ok := ctx.Deadline(); !ok {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
-		defer cancel()
+func (m *clientManager) UnRegister(scheme string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if client, ok := m.clients[strings.ToLower(scheme)]; ok {
+		logger.Infof("remove client %#v for scheme %s", client, scheme)
 	}
-	return sourceClient.IsExpired(ctx, url, header, expireInfo)
+	delete(m.clients, strings.ToLower(scheme))
 }
 
-func (m *clientManager) Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return nil, err
-	}
-	return sourceClient.Download(ctx, url, header, rang)
+func (m *clientManager) GetClient(scheme string) (ResourceClient, bool) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	client, ok := m.clients[strings.ToLower(scheme)]
+	return client, ok
 }
 
-func (m *clientManager) DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader,
-	error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return nil, nil, err
-	}
-	return sourceClient.DownloadWithResponseHeader(ctx, url, header, rang)
+func Register(scheme string, resourceClient ResourceClient, adaptor requestAdapter, hooks ...Hook) error {
+	return _defaultManager.Register(scheme, resourceClient, adaptor, hooks...)
 }
 
-func (m *clientManager) GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	sourceClient, err := m.getSourceClient(url)
-	if err != nil {
-		return -1, err
-	}
-	if _, ok := ctx.Deadline(); !ok {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
-		defer cancel()
-	}
-	return sourceClient.GetLastModifiedMillis(ctx, url, header)
+func UnRegister(scheme string) {
+	_defaultManager.UnRegister(scheme)
 }
 
-func NewManager() ClientManager {
-	return &clientManager{
-		clients: make(map[string]ResourceClient),
-	}
+type requestAdapter func(request *Request) *Request
+
+// Hook TODO hook
+type Hook interface {
+	BeforeRequest(request *Request) error
+	AfterResponse(response *Response) error
 }
 
-func (m *clientManager) Register(schema string, resourceClient ResourceClient) {
-	if client, ok := m.clients[strings.ToLower(schema)]; ok {
-		logger.Infof("replace client %#v with %#v for schema %s", client, resourceClient, schema)
-	}
-	m.clients[strings.ToLower(schema)] = resourceClient
+type clientWrapper struct {
+	adapter requestAdapter
+	hooks   []Hook
+	rc      ResourceClient
 }
 
-func Register(schema string, resourceClient ResourceClient) {
-	_defaultManager.Register(schema, resourceClient)
+func (c *clientWrapper) GetContentLength(request *Request) (int64, error) {
+	return c.rc.GetContentLength(c.adapter(request))
 }
 
-func (m *clientManager) UnRegister(schema string) {
-	if client, ok := m.clients[strings.ToLower(schema)]; ok {
-		logger.Infof("remove client %#v for schema %s", client, schema)
-	}
-	delete(m.clients, strings.ToLower(schema))
+func (c *clientWrapper) IsSupportRange(request *Request) (bool, error) {
+	return c.rc.IsSupportRange(c.adapter(request))
 }
 
-func UnRegister(schema string) {
-	_defaultManager.UnRegister(schema)
+func (c *clientWrapper) IsExpired(request *Request, info *ExpireInfo) (bool, error) {
+	return c.rc.IsExpired(c.adapter(request), info)
+}
+func (c *clientWrapper) Download(request *Request) (io.ReadCloser, error) {
+	return c.rc.Download(c.adapter(request))
 }
 
-func GetContentLength(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	var httpRange *rangers.Range
-	if header.Get(headers.Range) != "" {
-		var err error
-		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
-		if err != nil {
-			return -1, err
-		}
-	}
-	return _defaultManager.GetContentLength(ctx, url, header, httpRange)
+func (c *clientWrapper) DownloadWithExpireInfo(request *Request) (io.ReadCloser, *ExpireInfo, error) {
+	return c.rc.DownloadWithExpireInfo(c.adapter(request))
 }
 
-func IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
-	return _defaultManager.IsSupportRange(ctx, url, header)
+func (c *clientWrapper) GetLastModified(request *Request) (int64, error) {
+	return c.rc.GetLastModified(c.adapter(request))
 }
 
-func IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error) {
-	return _defaultManager.IsExpired(ctx, url, header, expireInfo)
+func GetContentLength(request *Request) (int64, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return UnKnownSourceFileLen, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
+	}
+	if _, ok := request.Context().Deadline(); !ok {
+		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		request = request.WithContext(ctx)
+		defer cancel()
+	}
+	return client.GetContentLength(request)
 }
 
-func Download(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, error) {
-	var httpRange *rangers.Range
-	if header.Get(headers.Range) != "" {
-		var err error
-		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
-		if err != nil {
-			return nil, err
-		}
+func IsSupportRange(request *Request) (bool, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return false, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
+	}
+	if _, ok := request.Context().Deadline(); !ok {
+		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		request = request.WithContext(ctx)
+		defer cancel()
+	}
+	if request.Header.get(Range) == "" {
+		request.Header.Add(Range, "0-0")
+	}
+	return client.IsSupportRange(request)
+}
+
+func IsExpired(request *Request, info *ExpireInfo) (bool, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return false, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
+	}
+	if _, ok := request.Context().Deadline(); !ok {
+		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		request = request.WithContext(ctx)
+		defer cancel()
 	}
-	return _defaultManager.Download(ctx, url, header, httpRange)
+	return client.IsExpired(request, info)
 }
 
-func DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, ResponseHeader, error) {
+func GetLastModified(request *Request) (int64, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return -1, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
+	}
+	if _, ok := request.Context().Deadline(); !ok {
+		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		request = request.WithContext(ctx)
+		defer cancel()
+	}
+	return client.GetLastModified(request)
+}
 
-	var httpRange *rangers.Range
-	if header.Get(headers.Range) != "" {
-		var err error
-		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
-		if err != nil {
-			return nil, nil, err
-		}
+func Download(request *Request) (io.ReadCloser, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return nil, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
 	}
-	return _defaultManager.DownloadWithResponseHeader(ctx, url, header, httpRange)
+	return client.Download(request)
 }
 
-func GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	return _defaultManager.GetLastModifiedMillis(ctx, url, header)
+func DownloadWithExpireInfo(request *Request) (io.ReadCloser, *ExpireInfo, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return nil, nil, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
+	}
+	return client.DownloadWithExpireInfo(request)
 }
 
 // getSourceClient get a source client from source manager with specified schema.
@@ -225,9 +293,9 @@ func (m *clientManager) getSourceClient(rawURL string) (ResourceClient, error) {
 	if err != nil {
 		return nil, err
 	}
-	m.RLock()
+	m.mu.RLock()
 	client, ok := m.clients[strings.ToLower(parsedURL.Scheme)]
-	m.RUnlock()
+	m.mu.RUnlock()
 	if !ok || client == nil {
 		client, err = m.loadSourcePlugin(strings.ToLower(parsedURL.Scheme))
 		if err == nil && client != nil {
@@ -238,19 +306,19 @@ func (m *clientManager) getSourceClient(rawURL string) (ResourceClient, error) {
 	return client, nil
 }
 
-func (m *clientManager) loadSourcePlugin(schema string) (ResourceClient, error) {
-	m.Lock()
-	defer m.Unlock()
+func (m *clientManager) loadSourcePlugin(scheme string) (ResourceClient, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	// double check
-	client, ok := m.clients[schema]
+	client, ok := m.clients[scheme]
 	if ok {
 		return client, nil
 	}
 
-	client, err := LoadPlugin(schema)
+	client, err := LoadPlugin(scheme)
 	if err != nil {
 		return nil, err
 	}
-	m.clients[schema] = client
+	m.clients[scheme] = client
 	return client, nil
 }
diff --git a/pkg/util/rangeutils/range.go b/pkg/util/rangeutils/range.go
index 8aee0574..57e1b732 100644
--- a/pkg/util/rangeutils/range.go
+++ b/pkg/util/rangeutils/range.go
@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+
+	"github.com/pkg/errors"
 )
 
 const (
@@ -35,9 +37,9 @@ func (r Range) String() string {
 	return fmt.Sprintf("%d%s%d", r.StartIndex, separator, r.EndIndex)
 }
 
-// ParseRange parses Range according to range string.
+// GetRange parses Range according to range string.
 // rangeStr: "start-end"
-func ParseRange(rangeStr string) (r *Range, err error) {
+func GetRange(rangeStr string) (r *Range, err error) {
 	ranges := strings.Split(rangeStr, separator)
 	if len(ranges) != 2 {
 		return nil, fmt.Errorf("range value(%s) is illegal which should be like 0-45535", rangeStr)
@@ -62,52 +64,96 @@ func ParseRange(rangeStr string) (r *Range, err error) {
 	}, nil
 }
 
-// ParseRange parses Range according to range string.
-// rangeStr: "bytes=start-end"
-func ParseHTTPRange(rangeStr string) (r *Range, err error) {
-	prefix := "bytes="
-	if !strings.HasPrefix(rangeStr, prefix) || strings.Count(rangeStr, prefix) != 1 {
-		return nil, fmt.Errorf("rangeStr %s not start with bytes=", rangeStr)
+// ParseRange parses the start and the end from rangeStr and returns them.
+// length is file total length
+func ParseRange(rangeStr string, length uint64) (*Range, error) {
+	if strings.Count(rangeStr, "-") != 1 {
+		return nil, errors.Errorf("invalid range: %s, should be like 0-1023", rangeStr)
 	}
-	rangeStr = strings.Replace(rangeStr, prefix, "", 1)
-	ranges := strings.Split(rangeStr, separator)
-	if len(ranges) != 2 {
-		return nil, fmt.Errorf("range value(%s) is illegal which should be like 0-45535", rangeStr)
+
+	// -{endIndex}
+	if strings.HasPrefix(rangeStr, "-") {
+		rangeStruct, err := handlePrefixRange(rangeStr, length)
+		if err != nil {
+			return nil, err
+		}
+		return rangeStruct, nil
 	}
 
-	startIndex, err := strconv.ParseUint(ranges[0], 10, 64)
+	// {startIndex}-
+	if strings.HasSuffix(rangeStr, "-") {
+		rangeStruct, err := handleSuffixRange(rangeStr, length)
+		if err != nil {
+			return nil, err
+		}
+		return rangeStruct, nil
+	}
+
+	rangeStruct, err := handlePairRange(rangeStr, length)
 	if err != nil {
-		return nil, fmt.Errorf("range(%s) start is not a non-negative number", rangeStr)
+		return nil, err
 	}
-	endIndex, err := strconv.ParseUint(ranges[1], 10, 64)
+	return rangeStruct, nil
+}
+
+func handlePrefixRange(rangeStr string, length uint64) (*Range, error) {
+	downLength, err := strconv.ParseUint(strings.TrimPrefix(rangeStr, "-"), 10, 64)
 	if err != nil {
-		return nil, fmt.Errorf("range(%s) end is not a non-negative number", rangeStr)
+		return nil, errors.Errorf("failed to parse range: %s to int: %v", rangeStr, err)
 	}
 
-	if endIndex < startIndex {
-		return nil, fmt.Errorf("range(%s) start is larger than end", rangeStr)
+	if downLength > length {
+		return nil, errors.Errorf("range: %s, the downLength is larger than length", rangeStr)
+	}
+
+	return &Range{
+		StartIndex: length - downLength,
+		EndIndex:   length - 1,
+	}, nil
+}
+
+func handleSuffixRange(rangeStr string, length uint64) (*Range, error) {
+	startIndex, err := strconv.ParseUint(strings.TrimSuffix(rangeStr, "-"), 10, 64)
+	if err != nil {
+		return nil, errors.Errorf("failed to parse range: %s to uint: %v", rangeStr, err)
+	}
+
+	if startIndex > length {
+		return nil, errors.Errorf("range: %s, the startIndex is larger than length", rangeStr)
 	}
 
 	return &Range{
 		StartIndex: startIndex,
-		EndIndex:   endIndex,
+		EndIndex:   length - 1,
 	}, nil
 }
 
-func GetBreakRange(breakPoint int64, sourceFileLength int64) (*Range, error) {
-	if breakPoint < 0 {
-		return nil, fmt.Errorf("breakPoint is illegal for value: %d", breakPoint)
+func handlePairRange(rangeStr string, length uint64) (*Range, error) {
+	rangePair := strings.Split(rangeStr, "-")
+
+	startIndex, err := strconv.ParseUint(rangePair[0], 10, 64)
+	if err != nil {
+		return nil, errors.Errorf("failed to parse range: %s to uint: %v", rangeStr, err)
 	}
-	if sourceFileLength < 0 {
-		return nil, fmt.Errorf("sourceFileLength is illegal for value: %d", sourceFileLength)
+	if startIndex > length {
+		return nil, errors.Errorf("range: %s, the startIndex is larger than length", rangeStr)
 	}
-	end := sourceFileLength - 1
-	if breakPoint > end {
-		return nil, fmt.Errorf("start: %d is larger than end: %d", breakPoint, end)
 
+	endIndex, err := strconv.ParseUint(rangePair[1], 10, 64)
+	if err != nil {
+		return nil, errors.Errorf("failed to parse range: %s to uint: %v", rangeStr, err)
+	}
+	if endIndex >= length {
+		//attention
+		endIndex = length - 1
+	}
+
+	if endIndex < startIndex {
+		return nil, errors.Errorf("range: %s, the start is larger the end", rangeStr)
 	}
+
 	return &Range{
-		StartIndex: uint64(breakPoint),
-		EndIndex:   uint64(end),
+		StartIndex: startIndex,
+		EndIndex:   endIndex,
 	}, nil
 }
diff --git a/pkg/util/timeutils/time_utils.go b/pkg/util/timeutils/time_utils.go
index be86e110..9bcc6faa 100644
--- a/pkg/util/timeutils/time_utils.go
+++ b/pkg/util/timeutils/time_utils.go
@@ -22,6 +22,8 @@ import (
 	"time"
 )
 
+const timeFormat = "Mon, 02 Jan 2006 15:04:05 GMT"
+
 func CurrentTimeMillis() int64 {
 	return time.Now().UnixNano() / time.Millisecond.Nanoseconds()
 }
@@ -32,7 +34,7 @@ func SinceInMilliseconds(start time.Time) int64 {
 
 // UnixMillis converts a {Mon, 02 Jan 2006 15:04:05 GMT} time to an int64 milliseconds.
 func UnixMillis(timeString string) int64 {
-	t, err := time.ParseInLocation(http.TimeFormat, timeString, time.UTC)
+	t, err := time.ParseInLocation(timeFormat, timeString, time.UTC)
 	if err != nil {
 		return 0
 	}
@@ -40,7 +42,26 @@ func UnixMillis(timeString string) int64 {
 	return t.UnixNano() / time.Millisecond.Nanoseconds()
 }
 
+func Format(time time.Time) string {
+	return time.Format(timeFormat)
+}
+
 // MillisUnixTime converts an int64 milliseconds to a unixTime
 func MillisUnixTime(millis int64) time.Time {
 	return time.Unix(millis/1000, millis%1000*int64(time.Millisecond))
 }
+
+// UnixSeconds converts a {Mon, 02 Jan 2006 15:04:05 GMT} time to an int64 seconds.
+func UnixSeconds(timeString string) int64 {
+	t, err := time.ParseInLocation(http.TimeFormat, timeString, time.UTC)
+	if err != nil {
+		return 0
+	}
+
+	return t.Unix()
+}
+
+// SecondsUnixTime converts an int64 seconds to a unixTime
+func SecondsUnixTime(seconds int64) time.Time {
+	return time.Unix(seconds, 0)
+}
diff --git a/scheduler/supervisor/mocks/host_mock.go b/scheduler/supervisor/mocks/host_mock.go
index 1175d864..833ac9db 100644
--- a/scheduler/supervisor/mocks/host_mock.go
+++ b/scheduler/supervisor/mocks/host_mock.go
@@ -1,3 +1,4 @@
+// Code generated by MockGen. DO NOT EDIT.
 // Source: d7y.io/dragonfly/v2/scheduler/supervisor (interfaces: HostManager)
 
 // Package mocks is a generated GoMock package.
@@ -6,9 +7,8 @@ package mocks
 import (
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
-
 	supervisor "d7y.io/dragonfly/v2/scheduler/supervisor"
+	gomock "github.com/golang/mock/gomock"
 )
 
 // MockHostManager is a mock of HostManager interface.


Test patch:
diff --git a/cdn/supervisor/cdn/cache_detector_test.go b/cdn/supervisor/cdn/cache_detector_test.go
index b38c580e..6d8b2e5b 100644
--- a/cdn/supervisor/cdn/cache_detector_test.go
+++ b/cdn/supervisor/cdn/cache_detector_test.go
@@ -22,6 +22,7 @@ import (
 	"hash"
 	"io"
 	"io/ioutil"
+	"os"
 	"sort"
 	"strings"
 	"testing"
@@ -36,6 +37,7 @@ import (
 	storageMock "d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage/mock"
 	"d7y.io/dragonfly/v2/cdn/types"
 	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
@@ -53,7 +55,8 @@ type CacheDetectorTestSuite struct {
 func (suite *CacheDetectorTestSuite) SetupSuite() {
 	ctrl := gomock.NewController(suite.T())
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	suite.Require().Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	storageMgr := storageMock.NewMockManager(ctrl)
 	cacheDataManager := newCacheDataManager(storageMgr)
 	suite.detector = newCacheDetector(cacheDataManager)
@@ -80,29 +83,30 @@ func (suite *CacheDetectorTestSuite) SetupSuite() {
 			suite.Nil(err)
 			return ioutil.NopCloser(strings.NewReader(string(content))), nil
 		}).AnyTimes()
-	storageMgr.EXPECT().ReadDownloadFile(noCache.taskID).Return(nil, cdnerrors.ErrFileNotExist{}).AnyTimes()
+	storageMgr.EXPECT().ReadDownloadFile(noCache.taskID).Return(nil, os.ErrNotExist).AnyTimes()
 	storageMgr.EXPECT().ReadPieceMetaRecords(fullNoExpiredCache.taskID).Return(fullNoExpiredCache.pieces, nil).AnyTimes()
 	storageMgr.EXPECT().ReadPieceMetaRecords(partialNotSupportRangeCache.taskID).Return(partialNotSupportRangeCache.pieces, nil).AnyTimes()
 	storageMgr.EXPECT().ReadPieceMetaRecords(partialSupportRangeCache.taskID).Return(partialSupportRangeCache.pieces, nil).AnyTimes()
-	storageMgr.EXPECT().ReadPieceMetaRecords(noCache.taskID).Return(nil, cdnerrors.ErrFileNotExist{}).AnyTimes()
+	storageMgr.EXPECT().ReadPieceMetaRecords(noCache.taskID).Return(nil, os.ErrNotExist).AnyTimes()
 	storageMgr.EXPECT().StatDownloadFile(fullNoExpiredCache.taskID).Return(&storedriver.StorageInfo{
 		Path:       "",
 		Size:       9789,
 		CreateTime: time.Time{},
 		ModTime:    time.Time{},
 	}, nil).AnyTimes()
+	storageMgr.EXPECT().StatDownloadFile(gomock.Not(fullNoExpiredCache.taskID)).Return(&storedriver.StorageInfo{}, nil).AnyTimes()
 
-	sourceClient.EXPECT().IsExpired(gomock.Any(), expiredAndSupportURL, gomock.Any(), gomock.Any()).Return(true, nil).AnyTimes()
-	sourceClient.EXPECT().IsSupportRange(gomock.Any(), expiredAndSupportURL, gomock.Any()).Return(true, nil).AnyTimes()
+	sourceClient.EXPECT().IsExpired(source.RequestEq(expiredAndSupportURL), gomock.Any()).Return(true, nil).AnyTimes()
+	sourceClient.EXPECT().IsSupportRange(source.RequestEq(expiredAndSupportURL)).Return(true, nil).AnyTimes()
 
-	sourceClient.EXPECT().IsExpired(gomock.Any(), expiredAndNotSupportURL, gomock.Any(), gomock.Any()).Return(true, nil).AnyTimes()
-	sourceClient.EXPECT().IsSupportRange(gomock.Any(), expiredAndNotSupportURL, gomock.Any()).Return(false, nil).AnyTimes()
+	sourceClient.EXPECT().IsExpired(source.RequestEq(noExpiredAndSupportURL), gomock.Any()).Return(false, nil).AnyTimes()
+	sourceClient.EXPECT().IsSupportRange(source.RequestEq(noExpiredAndSupportURL)).Return(true, nil).AnyTimes()
 
-	sourceClient.EXPECT().IsExpired(gomock.Any(), noExpiredAndSupportURL, gomock.Any(), gomock.Any()).Return(false, nil).AnyTimes()
-	sourceClient.EXPECT().IsSupportRange(gomock.Any(), noExpiredAndSupportURL, gomock.Any()).Return(true, nil).AnyTimes()
+	sourceClient.EXPECT().IsExpired(source.RequestEq(expiredAndNotSupportURL), gomock.Any()).Return(true, nil).AnyTimes()
+	sourceClient.EXPECT().IsSupportRange(source.RequestEq(expiredAndNotSupportURL)).Return(false, nil).AnyTimes()
 
-	sourceClient.EXPECT().IsExpired(gomock.Any(), noExpiredAndNotSupportURL, gomock.Any(), gomock.Any()).Return(false, nil).AnyTimes()
-	sourceClient.EXPECT().IsSupportRange(gomock.Any(), noExpiredAndNotSupportURL, gomock.Any()).Return(false, nil).AnyTimes()
+	sourceClient.EXPECT().IsExpired(source.RequestEq(noExpiredAndNotSupportURL), gomock.Any()).Return(false, nil).AnyTimes()
+	sourceClient.EXPECT().IsSupportRange(source.RequestEq(noExpiredAndNotSupportURL)).Return(false, nil).AnyTimes()
 }
 
 var noCacheTask, partialAndSupportCacheTask, partialAndNotSupportCacheTask, fullCacheExpiredTask, fullCacheNotExpiredTask = "noCache", "partialSupportCache",
diff --git a/cdn/supervisor/cdn/downloader_test.go b/cdn/supervisor/cdn/downloader_test.go
new file mode 100644
index 00000000..e07a49e8
--- /dev/null
+++ b/cdn/supervisor/cdn/downloader_test.go
@@ -0,0 +1,101 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cdn
+
+import "testing"
+
+func TestGetBreakRange(t *testing.T) {
+	type args struct {
+		breakPoint       int64
+		sourceFileLength int64
+		taskRange        string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "breakPoint with range",
+			args: args{
+				breakPoint:       5,
+				sourceFileLength: 200,
+				taskRange:        "3-103",
+			},
+			want:    "8-103",
+			wantErr: false,
+		}, {
+			name: "range with breakPoint 1",
+			args: args{
+				breakPoint:       1,
+				sourceFileLength: 100,
+				taskRange:        "0-100",
+			},
+			want:    "1-99",
+			wantErr: false,
+		}, {
+			name: "breakpoint larger than length of download required",
+			args: args{
+				breakPoint:       101,
+				sourceFileLength: 200,
+				taskRange:        "100-300",
+			},
+			want:    "",
+			wantErr: true,
+		}, {
+			name: "breakpoint is equal with length of download required",
+			args: args{
+				breakPoint:       100,
+				sourceFileLength: 200,
+				taskRange:        "100-300",
+			},
+			want:    "",
+			wantErr: true,
+		}, {
+			name: "breakpoint is smaller than length of download required",
+			args: args{
+				breakPoint:       99,
+				sourceFileLength: 200,
+				taskRange:        "100-300",
+			},
+			want:    "199-199",
+			wantErr: false,
+		}, {
+			name: "test2",
+			args: args{
+				breakPoint:       102760448,
+				sourceFileLength: 552562021,
+				taskRange:        "",
+			},
+			want:    "102760448-552562020",
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := getBreakRange(tt.args.breakPoint, tt.args.taskRange, tt.args.sourceFileLength)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("getBreakRange() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("getBreakRange() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/cdn/supervisor/cdn/manager_test.go b/cdn/supervisor/cdn/manager_test.go
index 364c6fe6..bc15da8f 100644
--- a/cdn/supervisor/cdn/manager_test.go
+++ b/cdn/supervisor/cdn/manager_test.go
@@ -36,6 +36,7 @@ import (
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/net/urlutils"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
@@ -84,41 +85,47 @@ func (suite *CDNManagerTestSuite) TearDownSuite() {
 func (suite *CDNManagerTestSuite) TestTriggerCDN() {
 	ctrl := gomock.NewController(suite.T())
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	suite.Require().Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().IsSupportRange(gomock.Any(), dragonflyURL, gomock.Any()).Return(true, nil).AnyTimes()
-	sourceClient.EXPECT().IsExpired(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).Return(false, nil).AnyTimes()
-	sourceClient.EXPECT().Download(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
+
+	sourceClient.EXPECT().IsSupportRange(gomock.Any()).Return(true, nil).AnyTimes()
+	sourceClient.EXPECT().IsExpired(gomock.Any(), gomock.Any()).Return(false, nil).AnyTimes()
+	sourceClient.EXPECT().Download(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
 			content, _ := ioutil.ReadFile("../../testdata/cdn/go.html")
-			if rang != nil {
-				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(rang.StartIndex), int64(rang.EndIndex))), nil
+			if request.Header.Get(source.Range) != "" {
+				parsed, _ := rangeutils.GetRange(request.Header.Get(source.Range))
+				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(parsed.StartIndex), int64(parsed.EndIndex))), nil
 			}
 			return ioutil.NopCloser(strings.NewReader(string(content))), nil
 		},
 	).AnyTimes()
-	sourceClient.EXPECT().DownloadWithResponseHeader(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
+	sourceClient.EXPECT().DownloadWithExpireInfo(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
 			content, _ := ioutil.ReadFile("../../testdata/cdn/go.html")
-			if rang != nil {
-				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(rang.StartIndex), int64(rang.EndIndex))),
-					map[string]string{
-						source.LastModified: "Sun, 06 Jun 2021 12:52:30 GMT",
-						source.ETag:         "etag",
+			if request.Header.Get(source.Range) != "" {
+				parsed, _ := rangeutils.GetRange(request.Header.Get(source.Range))
+				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(parsed.StartIndex), int64(parsed.EndIndex))),
+					&source.ExpireInfo{
+						LastModified: "Sun, 06 Jun 2021 12:52:30 GMT",
+						ETag:         "etag",
 					}, nil
 			}
-			return ioutil.NopCloser(strings.NewReader(string(content))), map[string]string{
-				source.LastModified: "Sun, 06 Jun 2021 12:52:30 GMT",
-				source.ETag:         "etag",
-			}, nil
+			return ioutil.NopCloser(strings.NewReader(string(content))),
+				&source.ExpireInfo{
+					LastModified: "Sun, 06 Jun 2021 12:52:30 GMT",
+					ETag:         "etag",
+				}, nil
 		},
 	).AnyTimes()
-	sourceClient.EXPECT().GetLastModifiedMillis(gomock.Any(), dragonflyURL, gomock.Any()).Return(
+	sourceClient.EXPECT().GetLastModified(gomock.Any()).Return(
 		timeutils.UnixMillis("Sun, 06 Jun 2021 12:52:30 GMT"), nil).AnyTimes()
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
-			if rang != nil {
-				return int64(rang.EndIndex-rang.StartIndex) + 1, nil
+	sourceClient.EXPECT().GetContentLength(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (int64, error) {
+			if request.Header.Get(source.Range) != "" {
+				parsed, _ := rangeutils.GetRange(request.Header.Get(source.Range))
+				return int64(parsed.EndIndex-parsed.StartIndex) + 1, nil
 			}
 			fileInfo, _ := os.Stat("../../testdata/cdn/go.html")
 			return fileInfo.Size(), nil
diff --git a/client/daemon/peer/peertask_file_test.go b/client/daemon/peer/peertask_file_test.go
index 1794b31b..1766a6a7 100644
--- a/client/daemon/peer/peertask_file_test.go
+++ b/client/daemon/peer/peertask_file_test.go
@@ -19,6 +19,7 @@ package peer
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
@@ -28,6 +29,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
+	testifyrequire "github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
@@ -36,12 +38,13 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
 func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 	assert := testifyassert.New(t)
+	require := testifyrequire.New(t)
 	ctrl := gomock.NewController(t)
 
 	testBytes, err := ioutil.ReadFile(test.File)
@@ -82,15 +85,22 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 	})
 
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	require.Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
-			return int64(len(testBytes)), nil
+	sourceClient.EXPECT().GetContentLength(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (int64, error) {
+			if request.URL.String() == url {
+				return int64(len(testBytes)), nil
+			}
+			return -1, fmt.Errorf("unexpect url: %s", request.URL.String())
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
-			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
+	sourceClient.EXPECT().Download(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
+			if request.URL.String() == url {
+				return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
+			}
+			return nil, fmt.Errorf("unexpect url: %s", request.URL.String())
 		})
 
 	ptm := &peerTaskManager{
@@ -159,6 +169,7 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 
 func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	assert := testifyassert.New(t)
+	require := testifyrequire.New(t)
 	ctrl := gomock.NewController(t)
 
 	testBytes, err := ioutil.ReadFile(test.File)
@@ -200,15 +211,21 @@ func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		})
 
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	require.Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader, rang *rangers.Range) (int64, error) {
-			return -1, nil
+	sourceClient.EXPECT().GetContentLength(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (int64, error) {
+			if request.URL.String() == url {
+				return -1, nil
+			}
+			return -1, fmt.Errorf("unexpect url: %s", request.URL.String())
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
-			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
+	sourceClient.EXPECT().Download(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
+			if request.URL.String() == url {
+				return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
+			}
+			return nil, fmt.Errorf("unexpect url: %s", request.URL.String())
 		})
 
 	ptm := &peerTaskManager{
diff --git a/client/daemon/peer/peertask_manager_test.go b/client/daemon/peer/peertask_manager_test.go
index 736f69e8..c11701a9 100644
--- a/client/daemon/peer/peertask_manager_test.go
+++ b/client/daemon/peer/peertask_manager_test.go
@@ -339,7 +339,9 @@ func TestPeerTaskManager_StartStreamPeerTask_BackSource(t *testing.T) {
 		taskID = "task-0"
 	)
 
-	source.Register("http", httpprotocol.NewHTTPSourceClient())
+	source.UnRegister("http")
+	assert.Nil(source.Register("http", httpprotocol.NewHTTPSourceClient(), httpprotocol.Adapter))
+	defer source.UnRegister("http")
 
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		n, err := w.Write(testBytes)
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index 3ec1aeea..9ebbb43d 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -32,6 +32,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/phayes/freeport"
 	testifyassert "github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
@@ -48,9 +49,9 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
 func setupBackSourcePartialComponents(ctrl *gomock.Controller, testBytes []byte, opt componentsOption) (
@@ -213,14 +214,15 @@ func TestStreamPeerTask_BackSource_Partial_WithContentLength(t *testing.T) {
 		})
 
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	require.Nil(t, source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (int64, error) {
 			return int64(len(testBytes)), nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any()).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
diff --git a/client/daemon/peer/peertask_stream_test.go b/client/daemon/peer/peertask_stream_test.go
index 43c4813a..9319163b 100644
--- a/client/daemon/peer/peertask_stream_test.go
+++ b/client/daemon/peer/peertask_stream_test.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
+	testifyrequire "github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
@@ -35,12 +36,13 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
 func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 	assert := testifyassert.New(t)
+	require := testifyrequire.New(t)
 	ctrl := gomock.NewController(t)
 
 	testBytes, err := ioutil.ReadFile(test.File)
@@ -78,15 +80,18 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 			return rc, rc, nil
 		})
 
+	request, err := source.NewRequest(url)
+	assert.Nil(err, "create request")
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	require.Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(source.RequestEq(request.URL.String())).DoAndReturn(
+		func(request *source.Request) (int64, error) {
 			return int64(len(testBytes)), nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(source.RequestEq(request.URL.String())).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
@@ -147,6 +152,7 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 
 func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	assert := testifyassert.New(t)
+	require := testifyrequire.New(t)
 	ctrl := gomock.NewController(t)
 
 	testBytes, err := ioutil.ReadFile(test.File)
@@ -185,14 +191,17 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		})
 
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
-	source.Register("http", sourceClient)
+	source.UnRegister("http")
+	require.Nil(source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
+	request, err := source.NewRequest(url)
+	assert.Nil(err, "create reqeust")
+	sourceClient.EXPECT().GetContentLength(source.RequestEq(request.URL.String())).DoAndReturn(
+		func(request *source.Request) (int64, error) {
 			return -1, nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(source.RequestEq(request.URL.String())).DoAndReturn(
+		func(request *source.Request) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
diff --git a/client/daemon/peer/piece_downloader_test.go b/client/daemon/peer/piece_downloader_test.go
index 0487b86e..e5f9ed9c 100644
--- a/client/daemon/peer/piece_downloader_test.go
+++ b/client/daemon/peer/piece_downloader_test.go
@@ -31,17 +31,22 @@ import (
 
 	"github.com/go-http-utils/headers"
 	testifyassert "github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/daemon/test"
 	"d7y.io/dragonfly/v2/client/daemon/upload"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
-	_ "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
+	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 )
 
 func TestPieceDownloader_DownloadPiece(t *testing.T) {
 	assert := testifyassert.New(t)
+	source.UnRegister("http")
+	require.Nil(t, source.Register("http", httpprotocol.NewHTTPSourceClient(), httpprotocol.Adapter))
+	defer source.UnRegister("http")
 	testData, err := ioutil.ReadFile(test.File)
 	assert.Nil(err, "load test file")
 	pieceDownloadTimeout := 30 * time.Second
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 49b707b0..df7a2e49 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -32,6 +32,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
@@ -48,7 +49,9 @@ import (
 func TestPieceManager_DownloadSource(t *testing.T) {
 	assert := testifyassert.New(t)
 	ctrl := gomock.NewController(t)
-	source.Register("http", httpprotocol.NewHTTPSourceClient())
+	source.UnRegister("http")
+	require.Nil(t, source.Register("http", httpprotocol.NewHTTPSourceClient(), httpprotocol.Adapter))
+	defer source.UnRegister("http")
 	testBytes, err := ioutil.ReadFile(test.File)
 	assert.Nil(err, "load test file")
 
diff --git a/client/dfget/dfget_test.go b/client/dfget/dfget_test.go
index 8b58ef31..261e1ebb 100644
--- a/client/dfget/dfget_test.go
+++ b/client/dfget/dfget_test.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/internal/idgen"
@@ -43,7 +44,9 @@ func Test_downloadFromSource(t *testing.T) {
 	content := idgen.UUIDString()
 
 	sourceClient := sourcemock.NewMockResourceClient(gomock.NewController(t))
-	source.Register("http", sourceClient)
+	require.Nil(t, source.Register("http", sourceClient, func(request *source.Request) *source.Request {
+		return request
+	}))
 	defer source.UnRegister("http")
 
 	cfg := &config.DfgetConfig{
@@ -51,8 +54,9 @@ func Test_downloadFromSource(t *testing.T) {
 		Output: output,
 		Digest: strings.Join([]string{digestutils.Sha256Hash.String(), digestutils.Sha256(content)}, ":"),
 	}
-
-	sourceClient.EXPECT().Download(context.Background(), cfg.URL, nil, nil).Return(ioutil.NopCloser(strings.NewReader(content)), nil)
+	request, err := source.NewRequest(cfg.URL)
+	assert.Nil(t, err)
+	sourceClient.EXPECT().Download(request).Return(ioutil.NopCloser(strings.NewReader(content)), nil)
 
 	err = downloadFromSource(context.Background(), cfg, nil)
 	assert.Nil(t, err)
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client_test.go b/pkg/source/hdfsprotocol/hdfs_source_client_test.go
index bb5db877..8bfdc062 100644
--- a/pkg/source/hdfsprotocol/hdfs_source_client_test.go
+++ b/pkg/source/hdfsprotocol/hdfs_source_client_test.go
@@ -1,5 +1,4 @@
 //+build linux
-
 /*
  *     Copyright 2020 The Dragonfly Authors
  *
@@ -19,8 +18,6 @@
 package hdfsprotocol
 
 import (
-	"context"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
@@ -30,13 +27,12 @@ import (
 
 	"github.com/agiledragon/gomonkey"
 	"github.com/colinmarc/hdfs/v2"
-	"github.com/go-http-utils/headers"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/cdn/supervisor/cdn"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 )
 
 var sourceClient source.ResourceClient
@@ -48,20 +44,20 @@ const (
 	hdfsExistFileContentLength      int64 = 12
 	hdfsExistFileContent                  = "Hello World\n"
 	hdfsExistFileLastModifiedMillis int64 = 1625218150000
-	hdfsExistFileLastModified             = "2021-07-02 09:29:10"
+	hdfsExistFileLastModified             = "Mon, 02 Jan 2006 15:04:05 GMT"
 	hdfsExistFileRangeStart         int64 = 3
 	hdfsExistFileRangeEnd           int64 = 10
 )
 
 const (
 	hdfsNotExistFileURL                 = "hdfs://127.0.0.1:9000/user/root/input/f3.txt"
-	hdfsNotExistFileContentLength int64 = -1
+	hdfsNotExistFileContentLength int64 = source.UnKnownSourceFileLen
 )
 
-var fakeHDFSClient *hdfs.Client = &hdfs.Client{}
+var fakeHDFSClient = &hdfs.Client{}
 
 func testBefore() {
-	sourceClient = NewHDFSSourceClient(func(p *hdfsSourceClient) {
+	sourceClient = newHDFSSourceClient(func(p *hdfsSourceClient) {
 		p.clientMap[hdfsExistFileHost] = fakeHDFSClient
 	})
 }
@@ -86,8 +82,11 @@ func TestGetContentLength_OK(t *testing.T) {
 
 	defer patch.Reset()
 
+	contentLengthRequest, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
+	contentLengthRequest.Header.Add(source.Range, "0-12")
 	// exist file
-	length, err := sourceClient.GetContentLength(context.Background(), hdfsExistFileURL, nil, &rangeutils.Range{EndIndex: 12})
+	length, err := sourceClient.GetContentLength(contentLengthRequest)
 	assert.Equal(t, hdfsExistFileContentLength, length)
 	assert.Nil(t, err)
 
@@ -103,9 +102,11 @@ func TestGetContentLength_Fail(t *testing.T) {
 	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
 
 	defer patch.Reset()
-
+	hdfsNotExistFileRequest, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
+	hdfsNotExistFileRequest.Header.Add(source.Range, "0-10")
 	// not exist file
-	length, err := sourceClient.GetContentLength(context.Background(), hdfsNotExistFileURL, nil, &rangeutils.Range{EndIndex: 10})
+	length, err := sourceClient.GetContentLength(hdfsNotExistFileRequest)
 	assert.Equal(t, hdfsNotExistFileContentLength, length)
 	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
 }
@@ -122,8 +123,9 @@ func TestIsSupportRange_FileExist(t *testing.T) {
 	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
 
 	defer patch.Reset()
-
-	supportRange, err := sourceClient.IsSupportRange(context.Background(), hdfsExistFileURL, nil)
+	request, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
+	supportRange, err := sourceClient.IsSupportRange(request)
 	assert.Equal(t, true, supportRange)
 	assert.Nil(t, err)
 }
@@ -138,20 +140,15 @@ func TestIsSupportRange_FileNotExist(t *testing.T) {
 
 	defer patch.Reset()
 
-	supportRange, err := sourceClient.IsSupportRange(context.Background(), hdfsNotExistFileURL, nil)
+	request, err := source.NewRequest(hdfsNotExistFileURL)
+	assert.Nil(t, err)
+	supportRange, err := sourceClient.IsSupportRange(request)
 	assert.Equal(t, false, supportRange)
 	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
 }
 
-//
-func TestIsExpired_NoHeader(t *testing.T) {
-	// header not have Last-Modified
-	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{})
-	assert.Equal(t, true, expired)
-	assert.Nil(t, err)
-}
 func TestIsExpired_LastModifiedExpired(t *testing.T) {
-	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	lastModified, _ := time.Parse(source.LastModifiedLayout, hdfsExistFileLastModified)
 	var info os.FileInfo = fakeHDFSFileInfo{
 		modtime: lastModified,
 	}
@@ -163,9 +160,12 @@ func TestIsExpired_LastModifiedExpired(t *testing.T) {
 
 	defer patch.Reset()
 
+	request, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
 	// header have Last-Modified
-	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{
-		headers.LastModified: "2020-01-01 00:00:00",
+	expired, err := sourceClient.IsExpired(request, &source.ExpireInfo{
+		LastModified: timeutils.Format(time.Date(2020, 01, 01, 0, 0, 0, 0, time.UTC)),
+		ETag:         "",
 	})
 	assert.Equal(t, true, expired)
 	assert.Nil(t, err)
@@ -173,7 +173,7 @@ func TestIsExpired_LastModifiedExpired(t *testing.T) {
 }
 
 func TestIsExpired_LastModifiedNotExpired(t *testing.T) {
-	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	lastModified, _ := time.Parse(source.LastModifiedLayout, hdfsExistFileLastModified)
 	var info os.FileInfo = fakeHDFSFileInfo{
 		modtime: lastModified,
 	}
@@ -185,9 +185,12 @@ func TestIsExpired_LastModifiedNotExpired(t *testing.T) {
 
 	defer patch.Reset()
 
+	request, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
 	// header have Last-Modified
-	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{
-		headers.LastModified: hdfsExistFileLastModified,
+	expired, err := sourceClient.IsExpired(request, &source.ExpireInfo{
+		LastModified: hdfsExistFileLastModified,
+		ETag:         "",
 	})
 	assert.Equal(t, false, expired)
 	assert.Nil(t, err)
@@ -219,9 +222,12 @@ func Test_Download_FileExist_ByRang(t *testing.T) {
 
 	rang := &rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
 	// exist file
-	download, err := sourceClient.Download(context.Background(), hdfsExistFileURL, source.RequestHeader{
-		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
-	}, rang)
+	request, err := source.NewRequestWithHeader(hdfsExistFileURL, map[string]string{
+		source.Range: rang.String(),
+	})
+	assert.Nil(t, err)
+
+	download, err := sourceClient.Download(request)
 	data, _ := ioutil.ReadAll(download)
 
 	assert.Equal(t, hdfsExistFileContent, string(data))
@@ -239,16 +245,19 @@ func TestDownload_FileNotExist(t *testing.T) {
 	defer patch.Reset()
 
 	rang := rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
+
+	request, err := source.NewRequestWithHeader(hdfsNotExistFileURL, map[string]string{
+		source.Range: rang.String(),
+	})
+	assert.Nil(t, err)
 	// not exist file
-	download, err := sourceClient.Download(context.Background(), hdfsNotExistFileURL, source.RequestHeader{
-		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
-	}, &rang)
+	download, err := sourceClient.Download(request)
 	assert.Nil(t, download)
 	assert.EqualError(t, err, "open /user/root/input/f3.txt: file does not exist")
 }
 
 func Test_DownloadWithResponseHeader_FileExist_ByRange(t *testing.T) {
-	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	lastModified, _ := time.Parse(source.LastModifiedLayout, hdfsExistFileLastModified)
 	var reader *hdfs.FileReader = &hdfs.FileReader{}
 	patches := gomonkey.NewPatches()
 	defer patches.Reset()
@@ -273,11 +282,12 @@ func Test_DownloadWithResponseHeader_FileExist_ByRange(t *testing.T) {
 	})
 
 	rang := rangeutils.Range{StartIndex: uint64(hdfsExistFileRangeStart), EndIndex: uint64(hdfsExistFileRangeEnd)}
-	body, responseHeader, err := sourceClient.DownloadWithResponseHeader(context.Background(), hdfsExistFileURL, source.RequestHeader{
-		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
-	}, &rang)
+	request, err := source.NewRequest(hdfsExistFileURL)
 	assert.Nil(t, err)
-	assert.Equal(t, hdfsExistFileLastModified, responseHeader.Get(source.LastModified))
+	request.Header.Add(source.Range, rang.String())
+	body, expireInfo, err := sourceClient.DownloadWithExpireInfo(request)
+	assert.Nil(t, err)
+	assert.Equal(t, hdfsExistFileLastModified, expireInfo.LastModified)
 
 	data, _ := ioutil.ReadAll(body)
 	assert.Equal(t, string(data), string([]byte(hdfsExistFileContent)[hdfsExistFileRangeStart:hdfsExistFileRangeEnd]))
@@ -290,16 +300,17 @@ func TestDownloadWithResponseHeader_FileNotExist(t *testing.T) {
 	defer patch.Reset()
 
 	rang := rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
-	body, responseHeader, err := sourceClient.DownloadWithResponseHeader(context.Background(), hdfsNotExistFileURL, source.RequestHeader{
-		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
-	}, &rang)
+	request, err := source.NewRequest(hdfsNotExistFileURL)
+	assert.Nil(t, err)
+	request.Header.Add(source.Range, rang.String())
+	reader, expireInfo, err := sourceClient.DownloadWithExpireInfo(request)
 	assert.EqualError(t, err, "open /user/root/input/f3.txt: file does not exist")
-	assert.Nil(t, responseHeader)
-	assert.Nil(t, body)
+	assert.Nil(t, reader)
+	assert.Nil(t, expireInfo)
 }
 
-func TestGetLastModifiedMillis_FileExist(t *testing.T) {
-	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+func TestGetLastModified_FileExist(t *testing.T) {
+	lastModified, _ := time.Parse(source.LastModifiedLayout, hdfsExistFileLastModified)
 	var info os.FileInfo = fakeHDFSFileInfo{
 		modtime: lastModified,
 	}
@@ -311,12 +322,15 @@ func TestGetLastModifiedMillis_FileExist(t *testing.T) {
 
 	defer patch.Reset()
 
-	lastModifiedMillis, err := sourceClient.GetLastModifiedMillis(context.Background(), hdfsExistFileURL, nil)
+	request, err := source.NewRequest(hdfsExistFileURL)
+	assert.Nil(t, err)
+
+	lastModifiedMillis, err := sourceClient.GetLastModified(request)
 	assert.Nil(t, err)
 	assert.Equal(t, hdfsExistFileLastModifiedMillis, lastModifiedMillis)
 }
 
-func TestGetLastModifiedMillis_FileNotExist(t *testing.T) {
+func TestGetLastModified_FileNotExist(t *testing.T) {
 	stubRet := []gomonkey.OutputCell{
 		{Values: gomonkey.Params{nil, errors.New("stat /user/root/input/f3.txt: file does not exist")}},
 	}
@@ -325,13 +339,16 @@ func TestGetLastModifiedMillis_FileNotExist(t *testing.T) {
 
 	defer patch.Reset()
 
-	lastModifiedMillis, err := sourceClient.GetLastModifiedMillis(context.Background(), hdfsNotExistFileURL, nil)
+	request, err := source.NewRequest(hdfsNotExistFileURL)
+	assert.Nil(t, err)
+
+	lastModifiedMillis, err := sourceClient.GetLastModified(request)
 	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
 	assert.Equal(t, hdfsNotExistFileContentLength, lastModifiedMillis)
 }
 
 func TestNewHDFSSourceClient(t *testing.T) {
-	client := NewHDFSSourceClient()
+	client := newHDFSSourceClient()
 	assert.NotNil(t, client)
 
 	options := make([]HDFSSourceClientOption, 0)
@@ -342,7 +359,7 @@ func TestNewHDFSSourceClient(t *testing.T) {
 	}
 	options = append(options, option)
 
-	newHDFSSourceClient := NewHDFSSourceClient(options...)
+	newHDFSSourceClient := newHDFSSourceClient(options...)
 
 	assert.IsType(t, &hdfsSourceClient{}, newHDFSSourceClient)
 
diff --git a/pkg/source/header_test.go b/pkg/source/header_test.go
new file mode 100644
index 00000000..7a6b0aa6
--- /dev/null
+++ b/pkg/source/header_test.go
@@ -0,0 +1,46 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestHeader_Clone(t *testing.T) {
+	tests := []struct {
+		name string
+		h    Header
+		want Header
+	}{
+		{
+			name: "",
+			h: Header{
+				"aaa": []string{"ddd"}, "fff": []string{"cccc"},
+			},
+			want: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			xx := tt.h.Clone()
+			xx["aaa"] = []string{"bbbb"}
+			fmt.Println(xx["aaa"])
+			fmt.Println(tt.h["aaa"])
+		})
+	}
+}
diff --git a/pkg/source/httpprotocol/http_source_client_test.go b/pkg/source/httpprotocol/http_source_client_test.go
index 584e4a9b..6eebc285 100644
--- a/pkg/source/httpprotocol/http_source_client_test.go
+++ b/pkg/source/httpprotocol/http_source_client_test.go
@@ -21,12 +21,12 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
-	"net/url"
 	"testing"
 	"time"
 
 	"github.com/go-http-utils/headers"
 	"github.com/jarcoal/httpmock"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/suite"
 
 	"d7y.io/dragonfly/v2/pkg/source"
@@ -39,11 +39,11 @@ func TestHTTPSourceClientTestSuite(t *testing.T) {
 
 type HTTPSourceClientTestSuite struct {
 	suite.Suite
-	source.ResourceClient
+	httpClient *httpSourceClient
 }
 
 func (suite *HTTPSourceClientTestSuite) SetupSuite() {
-	suite.ResourceClient = NewHTTPSourceClient()
+	suite.httpClient = newHTTPSourceClient()
 	httpmock.ActivateNonDefault(_defaultHTTPClient)
 }
 
@@ -52,71 +52,86 @@ func (suite *HTTPSourceClientTestSuite) TearDownSuite() {
 }
 
 var (
-	timeoutURL               = "http://timeout.com"
-	normalURL                = "http://normal.com"
-	errorURL                 = "http://error.com"
-	forbiddenURL             = "http://forbidden.com"
-	notfoundURL              = "http://notfound.com"
-	normalNotSupportRangeURL = "http://notsuppertrange.com"
+	timeoutRawURL               = "https://timeout.com"
+	normalRawURL                = "https://normal.com"
+	expireRawURL                = "https://expired.com"
+	errorRawURL                 = "https://error.com"
+	forbiddenRawURL             = "https://forbidden.com"
+	notfoundRawURL              = "https://notfound.com"
+	normalNotSupportRangeRawURL = "https://notsuppertrange.com"
 )
 
 var (
-	testContent  = "l am test case"
-	lastModified = "Sun, 06 Jun 2021 12:52:30 GMT"
-	//etag         = "UMiJT4h7MCEAEgnqCLA2CdAaABnK" // todo etag business code can not obtain
-	etag = ""
+	testContent        = "l am test case"
+	lastModified       = "Sun, 06 Jun 2021 12:52:30 GMT"
+	expireLastModified = "Sun, 06 Jun 2021 11:52:30 GMT"
+	etag               = "UMiJT4h7MCEAEgnqCLA2CdAaABnK"
+	expireEtag         = "UMiJ2T4h7MCEAEgnqCLA2CdAaABnK"
 )
 
 func (suite *HTTPSourceClientTestSuite) SetupTest() {
 	httpmock.Reset()
-	httpmock.RegisterResponder(http.MethodGet, timeoutURL, func(request *http.Request) (*http.Response, error) {
+	httpmock.RegisterResponder(http.MethodGet, timeoutRawURL, func(request *http.Request) (*http.Response, error) {
 		// To simulate the timeout
 		time.Sleep(5 * time.Second)
 		return httpmock.NewStringResponse(http.StatusOK, "ok"), nil
 	})
 
-	httpmock.RegisterResponder(http.MethodGet, normalURL, func(request *http.Request) (*http.Response, error) {
+	httpmock.RegisterResponder(http.MethodGet, normalRawURL, func(request *http.Request) (*http.Response, error) {
 		if rang := request.Header.Get(headers.Range); rang != "" {
-			r, _ := rangeutils.ParseRange(rang[6:])
+			r, _ := rangeutils.GetRange(rang[6:])
+			header := http.Header{}
+			header.Set(headers.LastModified, lastModified)
+			header.Set(headers.ETag, etag)
 			res := &http.Response{
 				StatusCode:    http.StatusPartialContent,
 				ContentLength: int64(r.EndIndex) - int64(r.StartIndex) + int64(1),
 				Body:          httpmock.NewRespBodyFromString(testContent[r.StartIndex:r.EndIndex]),
-				Header: http.Header{
-					headers.LastModified: []string{lastModified},
-					headers.ETag:         []string{etag},
-				},
+				Header:        header,
 			}
 			return res, nil
 		}
 		if expire := request.Header.Get(headers.IfModifiedSince); expire != "" {
+			header := http.Header{}
+			header.Set(headers.LastModified, lastModified)
+			header.Set(headers.ETag, etag)
 			res := &http.Response{
 				StatusCode:    http.StatusNotModified,
 				ContentLength: int64(len(testContent)),
 				Body:          httpmock.NewRespBodyFromString(testContent),
-				Header: http.Header{
-					headers.LastModified: []string{lastModified},
-					headers.ETag:         []string{etag},
-				},
+				Header:        header,
 			}
 			return res, nil
 		}
+		header := http.Header{}
+		header.Set(headers.LastModified, lastModified)
+		header.Set(headers.ETag, etag)
 		res := &http.Response{
 			StatusCode:    http.StatusOK,
 			ContentLength: 14,
 			Body:          httpmock.NewRespBodyFromString(testContent),
-			Header: http.Header{
-				headers.LastModified: []string{lastModified},
-				headers.ETag:         []string{etag},
-			},
+			Header:        header,
 		}
 		return res, nil
 	})
 
-	httpmock.RegisterResponder(http.MethodGet, forbiddenURL, httpmock.NewStringResponder(http.StatusForbidden, "forbidden"))
-	httpmock.RegisterResponder(http.MethodGet, notfoundURL, httpmock.NewStringResponder(http.StatusNotFound, "not found"))
-	httpmock.RegisterResponder(http.MethodGet, normalNotSupportRangeURL, httpmock.NewStringResponder(http.StatusOK, testContent))
-	httpmock.RegisterResponder(http.MethodGet, errorURL, httpmock.NewErrorResponder(fmt.Errorf("error")))
+	header := http.Header{}
+	header.Set(headers.LastModified, lastModified)
+	header.Set(headers.ETag, etag)
+	httpmock.RegisterResponder(http.MethodGet, expireRawURL, func(request *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode:    http.StatusOK,
+			ContentLength: 14,
+			Body:          httpmock.NewRespBodyFromString(testContent),
+			Header:        header,
+		}, nil
+	})
+
+	httpmock.RegisterResponder(http.MethodGet, forbiddenRawURL, httpmock.NewStringResponder(http.StatusForbidden, "forbidden"))
+	httpmock.RegisterResponder(http.MethodGet, notfoundRawURL, httpmock.NewStringResponder(http.StatusNotFound, "not found"))
+	httpmock.RegisterResponder(http.MethodGet, normalNotSupportRangeRawURL, httpmock.NewStringResponder(http.StatusOK, testContent))
+	httpmock.RegisterResponder(http.MethodGet, errorRawURL, httpmock.NewErrorResponder(fmt.Errorf("error")))
+
 }
 
 func (suite *HTTPSourceClientTestSuite) TestNewHTTPSourceClient() {
@@ -132,114 +147,95 @@ func (suite *HTTPSourceClientTestSuite) TestNewHTTPSourceClient() {
 }
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientDownloadWithResponseHeader() {
-	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
-	reader, responseHeader, err := suite.DownloadWithResponseHeader(ctx, timeoutURL, source.RequestHeader{}, nil)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	timeoutRequest, err := source.NewRequestWithContext(ctx, timeoutRawURL, nil)
+	suite.Nil(err)
+	reader, expireInfo, err := suite.httpClient.DownloadWithExpireInfo(timeoutRequest)
 	cancel()
 	suite.NotNil(err)
-	suite.Equal("Get \"http://timeout.com\": context deadline exceeded", err.Error())
+	suite.Equal("Get \"https://timeout.com\": context deadline exceeded", err.Error())
 	suite.Nil(reader)
-	suite.Nil(responseHeader)
+	suite.Nil(expireInfo)
 
-	type args struct {
-		ctx    context.Context
-		url    string
-		header source.RequestHeader
-	}
+	normalRequest, _ := source.NewRequest(normalRawURL)
+	normalRangeRequest, _ := source.NewRequest(normalRawURL)
+	normalRangeRequest.Header.Add(headers.Range, fmt.Sprintf("bytes=%s", "0-3"))
+	notfoundRequest, _ := source.NewRequest(notfoundRawURL)
+	errorRequest, _ := source.NewRequest(errorRawURL)
 	tests := []struct {
 		name       string
-		args       args
+		request    *source.Request
 		content    string
-		expireInfo source.ResponseHeader
+		expireInfo *source.ExpireInfo
 		wantErr    error
 	}{
 		{
-			name: "normal download",
-			args: args{
-				ctx:    context.Background(),
-				url:    normalURL,
-				header: nil,
-			},
+			name:    "normal download",
+			request: normalRequest,
 			content: testContent,
-			expireInfo: source.ResponseHeader{
-				source.LastModified: lastModified,
-				source.ETag:         etag,
+			expireInfo: &source.ExpireInfo{
+				LastModified: lastModified,
+				ETag:         etag,
 			},
 			wantErr: nil,
 		}, {
-			name: "range download",
-			args: args{
-				ctx:    context.Background(),
-				url:    normalURL,
-				header: source.RequestHeader{"Range": fmt.Sprintf("bytes=%s", "0-3")},
-			},
+			name:    "range download",
+			request: normalRangeRequest,
 			content: testContent[0:3],
-			expireInfo: source.ResponseHeader{
-				headers.LastModified: lastModified,
-				headers.ETag:         etag,
+			expireInfo: &source.ExpireInfo{
+				LastModified: lastModified,
+				ETag:         etag,
 			},
 			wantErr: nil,
 		}, {
-			name: "not found download",
-			args: args{
-				ctx:    context.Background(),
-				url:    notfoundURL,
-				header: nil,
-			},
+			name:       "not found download",
+			request:    notfoundRequest,
 			content:    "",
 			expireInfo: nil,
-			wantErr:    fmt.Errorf("unexpected status code: %d", http.StatusNotFound),
+			wantErr:    source.CheckResponseCode(404, []int{200, 206}),
 		}, {
-			name: "error download",
-			args: args{
-				ctx:    context.Background(),
-				url:    errorURL,
-				header: nil,
-			},
+			name:       "error download",
+			request:    errorRequest,
 			content:    "",
 			expireInfo: nil,
-			wantErr: &url.Error{
-				Op:  "Get",
-				URL: errorURL,
-				Err: fmt.Errorf("error"),
-			},
+			wantErr:    errors.Errorf("Get \"https://error.com\": error"),
 		},
 	}
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			reader, responseHeader, err := suite.DownloadWithResponseHeader(tt.args.ctx, tt.args.url, tt.args.header, nil)
-			suite.Equal(tt.wantErr, err)
+			reader, expireInfo, err := suite.httpClient.DownloadWithExpireInfo(tt.request)
 			if err != nil {
+				suite.True(tt.wantErr.Error() == err.Error())
 				return
 			}
 			bytes, err := ioutil.ReadAll(reader)
 			suite.Nil(err)
 			suite.Equal(tt.content, string(bytes))
-			suite.Equal(tt.expireInfo, responseHeader)
+			suite.Equal(tt.expireInfo, expireInfo)
 		})
 	}
 }
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientGetContentLength() {
-	type args struct {
-		ctx    context.Context
-		url    string
-		header source.RequestHeader
-	}
+	normalRequest, _ := source.NewRequest(normalRawURL)
+	normalRangeRequest, _ := source.NewRequest(normalRawURL)
+	normalRangeRequest.Header.Add(headers.Range, fmt.Sprintf("bytes=%s", "0-3"))
 	tests := []struct {
 		name    string
-		args    args
+		request *source.Request
 		want    int64
 		wantErr error
 	}{
-		{name: "support content length", args: args{ctx: context.Background(), url: normalURL, header: map[string]string{}}, want: int64(len(testContent)),
+		{name: "support content length", request: normalRequest,
+			want:    int64(len(testContent)),
 			wantErr: nil},
-		{name: "not support content length", args: args{ctx: context.Background(), url: normalURL, header: source.RequestHeader{"Range": fmt.Sprintf("bytes=%s",
-			"0-3")}}, want: 4,
+		{name: "not support content length", request: normalRangeRequest,
+			want:    4,
 			wantErr: nil},
 	}
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			got, err := suite.GetContentLength(tt.args.ctx, tt.args.url, tt.args.header, nil)
+			got, err := suite.httpClient.GetContentLength(tt.request)
 			suite.Equal(tt.wantErr, err)
 			suite.Equal(tt.want, got)
 		})
@@ -247,27 +243,32 @@ func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientGetContentLength() {
 }
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientIsExpired() {
-	type args struct {
-		ctx        context.Context
-		url        string
-		header     source.RequestHeader
-		expireInfo map[string]string
-	}
+	normalRequest, _ := source.NewRequest(normalRawURL)
+	errorRequest, _ := source.NewRequest(errorRawURL)
+	expireRequest, _ := source.NewRequest(expireRawURL)
 	tests := []struct {
-		name    string
-		args    args
-		want    bool
-		wantErr bool
+		name       string
+		request    *source.Request
+		expireInfo *source.ExpireInfo
+		want       bool
+		wantErr    bool
 	}{
-		{name: "not expire", args: args{context.Background(), normalURL, source.RequestHeader{}, map[string]string{headers.LastModified: lastModified,
-			headers.ETag: etag}}, want: false, wantErr: false},
-		{name: "error not expire", args: args{context.Background(), errorURL, source.RequestHeader{}, map[string]string{headers.LastModified: lastModified,
-			headers.ETag: etag}}, want: false, wantErr: true},
-		{name: "expired", args: args{context.Background(), normalURL, source.RequestHeader{}, map[string]string{}}, want: true, wantErr: false},
+		{name: "not expire", request: normalRequest, expireInfo: &source.ExpireInfo{
+			LastModified: lastModified,
+			ETag:         etag,
+		}, want: false, wantErr: false},
+		{name: "error not expire", request: errorRequest, expireInfo: &source.ExpireInfo{
+			LastModified: lastModified,
+			ETag:         etag,
+		}, want: false, wantErr: true},
+		{name: "expired", request: expireRequest, expireInfo: &source.ExpireInfo{
+			LastModified: expireLastModified,
+			ETag:         expireEtag,
+		}, want: true, wantErr: false},
 	}
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			got, err := suite.IsExpired(tt.args.ctx, tt.args.url, tt.args.header, tt.args.expireInfo)
+			got, err := suite.httpClient.IsExpired(tt.request, tt.expireInfo)
 			suite.Equal(tt.want, got)
 			suite.Equal(tt.wantErr, err != nil)
 		})
@@ -275,39 +276,42 @@ func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientIsExpired() {
 }
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientIsSupportRange() {
-	httpmock.RegisterResponder(http.MethodGet, timeoutURL, func(request *http.Request) (*http.Response, error) {
+	httpmock.RegisterResponder(http.MethodGet, timeoutRawURL, func(request *http.Request) (*http.Response, error) {
 		time.Sleep(3 * time.Second)
 		return httpmock.NewStringResponse(http.StatusOK, "ok"), nil
 	})
 	parent := context.Background()
 	ctx, cancel := context.WithTimeout(parent, 1*time.Second)
-	support, err := suite.IsSupportRange(ctx, timeoutURL, nil)
+	request, err := source.NewRequestWithContext(ctx, timeoutRawURL, nil)
+	suite.Nil(err)
+	support, err := suite.httpClient.IsSupportRange(request)
 	cancel()
 	suite.NotNil(err)
-	suite.Equal("Get \"http://timeout.com\": context deadline exceeded", err.Error())
+	suite.Equal("Get \"https://timeout.com\": context deadline exceeded", err.Error())
 	suite.Equal(false, support)
-	httpmock.RegisterResponder(http.MethodGet, normalURL, httpmock.NewStringResponder(http.StatusPartialContent, ""))
-	httpmock.RegisterResponder(http.MethodGet, "http://notSupportRange.com", httpmock.NewStringResponder(http.StatusOK, ""))
-	httpmock.RegisterResponder(http.MethodGet, "http://error.com", httpmock.NewErrorResponder(fmt.Errorf("xxx")))
-	type args struct {
-		ctx    context.Context
-		url    string
-		header map[string]string
-	}
+	httpmock.RegisterResponder(http.MethodGet, normalRawURL, httpmock.NewStringResponder(http.StatusPartialContent, ""))
+	httpmock.RegisterResponder(http.MethodGet, normalNotSupportRangeRawURL, httpmock.NewStringResponder(http.StatusOK, ""))
+	httpmock.RegisterResponder(http.MethodGet, errorRawURL, httpmock.NewErrorResponder(fmt.Errorf("xxx")))
+
+	supportRangeRequest, _ := source.NewRequest(normalRawURL)
+	supportRangeRequest.Header.Add(headers.Range, fmt.Sprintf("bytes=%s", "0-3"))
+	notSupportRangeURL, _ := source.NewRequest(normalNotSupportRangeRawURL)
+	notSupportRangeURL.Header.Add(headers.Range, fmt.Sprintf("bytes=%s", "0-3"))
+	errRequest, _ := source.NewRequest(errorRawURL)
 	tests := []struct {
 		name    string
-		args    args
+		request *source.Request
 		want    bool
 		wantErr bool
 	}{
-		{name: "support", args: args{ctx: context.Background(), url: normalURL, header: source.RequestHeader{"Range": fmt.Sprintf("bytes=%s",
-			"0-3")}}, want: true, wantErr: false},
-		{name: "notSupport", args: args{ctx: context.Background(), url: normalNotSupportRangeURL, header: source.RequestHeader{"Range": fmt.Sprintf("bytes=%s",
-			"0-3")}}, want: false, wantErr: false},
+		{name: "support", request: supportRangeRequest, want: true, wantErr: false},
+		{name: "notSupport", request: notSupportRangeURL, want: false, wantErr: false},
+		{name: "error", request: errRequest, want: false, wantErr: true},
 	}
+
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			got, err := suite.IsSupportRange(tt.args.ctx, tt.args.url, tt.args.header)
+			got, err := suite.httpClient.IsSupportRange(tt.request)
 			suite.Equal(tt.wantErr, err != nil)
 			suite.Equal(tt.want, got)
 		})
@@ -315,9 +319,11 @@ func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientIsSupportRange() {
 }
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientDoRequest() {
-	var testURL = "http://www.hackhttp.com"
+	var testURL = "https://www.hackhttp.com"
 	httpmock.RegisterResponder(http.MethodGet, testURL, httpmock.NewStringResponder(http.StatusOK, "ok"))
-	res, err := suite.ResourceClient.(*httpSourceClient).doRequest(context.Background(), http.MethodGet, "http://www.hackhttp.com", nil)
+	request, err := source.NewRequest(testURL)
+	suite.Nil(err)
+	res, err := suite.httpClient.doRequest(http.MethodGet, request)
 	suite.Nil(err)
 	bytes, err := ioutil.ReadAll(res.Body)
 	suite.Nil(err)
diff --git a/pkg/source/request_test.go b/pkg/source/request_test.go
new file mode 100644
index 00000000..3c8e1185
--- /dev/null
+++ b/pkg/source/request_test.go
@@ -0,0 +1,86 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"context"
+	"net/url"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func equalRequestWithOutContext(expected, actual *Request) bool {
+	return expected.URL.String() == actual.URL.String() && reflect.DeepEqual(expected.Header, actual.Header)
+}
+
+func TestNewRequest(t *testing.T) {
+	assert := assert.New(t)
+	testURL, err := url.Parse("http://www.dragonfly.io")
+	assert.Nil(err)
+	got, err := NewRequest("http://www.dragonfly.io")
+	assert.Nil(err)
+	// test NewRequest
+	assert.True(equalRequestWithOutContext(&Request{URL: testURL,
+		Header: Header{}}, got))
+	testHeaderMap := map[string]string{"testKey1": "testValue1", "testKey2": "testValue2"}
+	testHeader := Header{"Testkey1": []string{"testValue1"}, "Testkey2": []string{"testValue2"}}
+	// test NewRequestWithHeader
+	got, err = NewRequestWithHeader("http://www.dragonfly.io", testHeaderMap)
+	assert.Nil(err)
+	assert.True(equalRequestWithOutContext(&Request{URL: testURL,
+		Header: testHeader}, got))
+
+	// test NewRequestWithContext
+	testContext, cancelFunc := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancelFunc()
+	got, err = NewRequestWithContext(testContext, "http://www.dragonfly.io", testHeaderMap)
+	assert.Nil(err)
+	contextRequest := &Request{
+		URL:    testURL,
+		Header: testHeader,
+		ctx:    testContext,
+	}
+	assert.True(equalRequestWithOutContext(contextRequest, got))
+	expectedDeadline, _ := testContext.Deadline()
+	gotDeadline, _ := got.ctx.Deadline()
+	assert.Equal(expectedDeadline, gotDeadline)
+}
+
+func TestRequest_Clone(t *testing.T) {
+	var testURL, err = url.Parse("http://www.dragonfly.io")
+	testCloneURL, err := url.Parse("http://www.dragonfly.io")
+	testHeader := Header{"testKey1": []string{"testValue1"}, "testKey2": []string{"testValue2"}}
+	testCloneHeader := Header{"testKey1": []string{"testValue1"}, "testKey2": []string{"testValue2"}}
+	testContext := context.Background()
+	require := require.New(t)
+	require.Nil(err)
+	r := &Request{
+		URL:    testURL,
+		Header: testHeader,
+		ctx:    testContext,
+	}
+	got := r.Clone(testContext)
+	require.EqualValues(&Request{
+		URL:    testCloneURL,
+		Header: testCloneHeader,
+		ctx:    testContext,
+	}, got)
+}
diff --git a/pkg/source/testdata/main.go b/pkg/source/testdata/main.go
index 0ef6019c..55b3d853 100644
--- a/pkg/source/testdata/main.go
+++ b/pkg/source/testdata/main.go
@@ -17,7 +17,6 @@
 package main
 
 import (
-	"context"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -40,15 +39,15 @@ func main() {
 		os.Exit(1)
 	}
 
-	ctx := context.Background()
-
-	l, err := client.GetContentLength(ctx, "", nil, nil)
+	request, err := source.NewRequest("")
+	l, err := client.GetContentLength(request)
 	if err != nil {
 		fmt.Printf("get content length error: %s\n", err)
 		os.Exit(1)
 	}
 
-	rc, err := client.Download(ctx, "", nil, nil)
+	request, err = source.NewRequest("")
+	rc, err := client.Download(request)
 	if err != nil {
 		fmt.Printf("download error: %s\n", err)
 		os.Exit(1)
diff --git a/pkg/source/testdata/plugin/dfs.go b/pkg/source/testdata/plugin/dfs.go
index 5de6f779..f74751cd 100644
--- a/pkg/source/testdata/plugin/dfs.go
+++ b/pkg/source/testdata/plugin/dfs.go
@@ -18,12 +18,9 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"io"
 	"io/ioutil"
 
-	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
-
 	"d7y.io/dragonfly/v2/pkg/source"
 )
 
@@ -34,30 +31,30 @@ var _ source.ResourceClient = (*client)(nil)
 type client struct {
 }
 
-func (c *client) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
+func (c *client) GetContentLength(request *source.Request) (int64, error) {
 	return int64(len(data)), nil
 }
 
-func (c *client) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
+func (c *client) IsSupportRange(request *source.Request) (bool, error) {
 	return false, nil
 }
 
-func (c *client) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
+func (c *client) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
 	panic("implement me")
 }
 
-func (c *client) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
+func (c *client) Download(request *source.Request) (io.ReadCloser, error) {
 	return ioutil.NopCloser(bytes.NewBufferString(data)), nil
 }
 
-func (c *client) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
-	return ioutil.NopCloser(bytes.NewBufferString(data)), map[string]string{}, nil
+func (c *client) DownloadWithExpireInfo(request *source.Request) (io.ReadCloser, *source.ExpireInfo, error) {
+	return ioutil.NopCloser(bytes.NewBufferString(data)), nil, nil
 }
 
-func (c *client) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+func (c *client) GetLastModified(request *source.Request) (int64, error) {
 	panic("implement me")
 }
 
 func DragonflyPluginInit(option map[string]string) (interface{}, map[string]string, error) {
-	return &client{}, map[string]string{"type": "resource", "name": "dfs", "schema": "dfs"}, nil
+	return &client{}, map[string]string{"type": "resource", "name": "dfs", "scheme": "dfs"}, nil
 }
diff --git a/pkg/util/rangeutils/range_test.go b/pkg/util/rangeutils/range_test.go
new file mode 100644
index 00000000..8b458c8b
--- /dev/null
+++ b/pkg/util/rangeutils/range_test.go
@@ -0,0 +1,104 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rangeutils
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseRange(t *testing.T) {
+	var cases = []struct {
+		rangeStr string
+		length   uint64
+		expected *Range
+		wantErr  bool
+	}{
+		{
+			rangeStr: "0-65575",
+			length:   65576,
+			expected: &Range{
+				StartIndex: 0,
+				EndIndex:   65575,
+			},
+			wantErr: false,
+		}, {
+			rangeStr: "2-2",
+			length:   65576,
+			expected: &Range{
+				StartIndex: 2,
+				EndIndex:   2,
+			},
+			wantErr: false,
+		}, {
+			rangeStr: "2-",
+			length:   65576,
+			expected: &Range{
+				StartIndex: 2,
+				EndIndex:   65575,
+			},
+			wantErr: false,
+		}, {
+			rangeStr: "-100",
+			length:   65576,
+			expected: &Range{
+				StartIndex: 65476,
+				EndIndex:   65575,
+			},
+			wantErr: false,
+		}, {
+			rangeStr: "0-66575",
+			length:   65576,
+			expected: &Range{
+				StartIndex: 0,
+				EndIndex:   65575,
+			},
+			wantErr: false,
+		},
+		{
+			rangeStr: "0-65-575",
+			length:   65576,
+			expected: nil,
+			wantErr:  true,
+		},
+		{
+			rangeStr: "0-hello",
+			length:   65576,
+			expected: nil,
+			wantErr:  true,
+		},
+		{
+			rangeStr: "65575-0",
+			length:   65576,
+			expected: nil,
+			wantErr:  true,
+		},
+		{
+			rangeStr: "-1-8",
+			length:   65576,
+			expected: nil,
+			wantErr:  true,
+		},
+	}
+
+	for _, v := range cases {
+		result, err := ParseRange(v.rangeStr, v.length)
+		assert.Equal(t, v.expected, result)
+		assert.Equal(t, v.wantErr, err != nil)
+	}
+}
diff --git a/pkg/util/timeutils/time_utils_test.go b/pkg/util/timeutils/time_utils_test.go
index 00541045..99c4c689 100644
--- a/pkg/util/timeutils/time_utils_test.go
+++ b/pkg/util/timeutils/time_utils_test.go
@@ -63,6 +63,32 @@ func TestUnixMillis(t *testing.T) {
 	}
 }
 
+func TestUnixSeconds(t *testing.T) {
+	const Layout = "Mon, 02 Jan 2006 15:04:05 GMT"
+	sample, _ := time.Parse(Layout, "Mon, 02 Jan 2006 15:04:05 GMT")
+	type args struct {
+		timeString string
+	}
+	tests := []struct {
+		name string
+		args args
+		want int64
+	}{
+		{
+			name: "convert a string time to an int64 milliseconds",
+			args: args{"Mon, 02 Jan 2006 15:04:05 GMT"},
+			want: sample.Unix(),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := UnixSeconds(tt.args.timeString); got != tt.want {
+				t.Errorf("UnixMillis() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestMillisUnixTime(t *testing.T) {
 	const Layout = "2006-01-02 15:04:05"
 	sample, _ := time.ParseInLocation(Layout, "2021-01-02 12:04:05", time.Local)
@@ -90,3 +116,31 @@ func TestMillisUnixTime(t *testing.T) {
 		})
 	}
 }
+
+func TestSecondsUnixTime(t *testing.T) {
+	const Layout = "2006-01-02 15:04:05"
+	sample, _ := time.ParseInLocation(Layout, "2021-01-02 12:04:05", time.Local)
+	tests := []struct {
+		name string
+		args int64
+		want time.Time
+	}{
+		{
+			name: "convert an int64 seconds to a unix time",
+			args: sample.Unix(),
+			want: sample.Local(),
+		},
+		{
+			name: "convert now",
+			args: time.Now().Unix(),
+			want: time.Unix(time.Now().Unix(), 0),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := SecondsUnixTime(tt.args); got != tt.want {
+				t.Errorf("MillisUnixTime() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}