func (factory *gardenFactory) Get(
	logger lager.Logger,
	stepMetadata StepMetadata,
	sourceName SourceName,
	id worker.Identifier,
	delegate GetDelegate,
	resourceConfig atc.ResourceConfig,
	params atc.Params,
	tags atc.Tags,
	version atc.Version,
) StepFactory {
	id.WorkingDirectory = resource.ResourcesDir("get")
	return newGetStep(
		logger,
		sourceName,
		resourceConfig,
		version,
		params,
		resource.ResourceCacheIdentifier{
			Type:    resource.ResourceType(resourceConfig.Type),
			Source:  resourceConfig.Source,
			Params:  params,
			Version: version,
		},
		stepMetadata,
		resource.Session{
			ID:        id,
			Ephemeral: false,
		},
		tags,
		delegate,
		factory.tracker,
	)
}

func (factory *gardenFactory) Put(id worker.Identifier, delegate PutDelegate, config atc.ResourceConfig, tags atc.Tags, params atc.Params) StepFactory {
	return resourceStep{
		Session: resource.Session{
			ID: id,
		},

		Delegate: delegate,

		Tracker: factory.resourceTracker,
		Type:    resource.ResourceType(config.Type),
		Tags:    tags,

		Action: func(r resource.Resource, s ArtifactSource, vi VersionInfo) resource.VersionedSource {
			return r.Put(resource.IOConfig{
				Stdout: delegate.Stdout(),
				Stderr: delegate.Stderr(),
			}, config.Source, params, resourceSource{s})
		},
	}
}

func (factory *gardenFactory) Get(sourceName SourceName, id worker.Identifier, delegate GetDelegate, config atc.ResourceConfig, params atc.Params, tags atc.Tags, version atc.Version) StepFactory {
	return resourceStep{
		SourceName: sourceName,

		Session: resource.Session{
			ID:        id,
			Ephemeral: false,
		},

		Delegate: delegate,

		Tracker: factory.resourceTracker,
		Type:    resource.ResourceType(config.Type),
		Tags:    tags,

		Action: func(r resource.Resource, s ArtifactSource, vi VersionInfo) resource.VersionedSource {
			return r.Get(resource.IOConfig{
				Stdout: delegate.Stdout(),
				Stderr: delegate.Stderr(),
			}, config.Source, params, version)
		},
	}
}

// Using constructs a GetStep that will fetch the version of the resource
// determined by the VersionInfo result of the previous step.
func (step DependentGetStep) Using(prev Step, repo *SourceRepository) Step {
	var info VersionInfo
	prev.Result(&info)

	return newGetStep(
		step.logger,
		step.sourceName,
		step.resourceConfig,
		info.Version,
		step.params,
		resource.ResourceCacheIdentifier{
			Type:    resource.ResourceType(step.resourceConfig.Type),
			Source:  step.resourceConfig.Source,
			Params:  step.params,
			Version: info.Version,
		},
		step.stepMetadata,
		step.session,
		step.tags,
		step.delegate,
		step.tracker,
		step.resourceTypes,
	).Using(prev, repo)
}

				_, sm, sid, typ, tags, sources, actualResourceTypes, delegate := fakeTracker.InitWithSourcesArgsForCall(0)
				Expect(sm).To(Equal(stepMetadata))
				Expect(sid).To(Equal(resource.Session{
					ID: worker.Identifier{
						ResourceID: 1234,
						Stage:      db.ContainerStageRun,
					},
					Metadata: worker.Metadata{
						PipelineName:     "some-pipeline",
						Type:             db.ContainerTypePut,
						StepName:         "some-step",
						WorkingDirectory: "/tmp/build/put",
					},
				}))
				Expect(typ).To(Equal(resource.ResourceType("some-resource-type")))
				Expect(tags).To(ConsistOf("some", "tags"))
				Expect(actualResourceTypes).To(Equal(atc.ResourceTypes{
					{
						Name:   "custom-resource",
						Type:   "custom-type",
						Source: atc.Source{"some-custom": "source"},
					},
				}))
				Expect(delegate).To(Equal(putDelegate))

				// TODO: Can we test the map values?
				Expect(sources).To(HaveKey("some-source"))
				Expect(sources).To(HaveKey("some-other-source"))
				Expect(sources).To(HaveKey("some-mounted-source"))
			})

				fakeVersionedSource.VersionReturns(atc.Version{"some": "version"})
				fakeVersionedSource.MetadataReturns([]atc.MetadataField{{"some", "metadata"}})

				fakeResource.GetReturns(fakeVersionedSource)
			})

			It("initializes the resource with the correct type and session id, making sure that it is not ephemeral", func() {
				Ω(fakeTracker.InitCallCount()).Should(Equal(1))

				sid, typ, tags := fakeTracker.InitArgsForCall(0)

				Ω(sid).Should(Equal(resource.Session{
					ID:        identifier,
					Ephemeral: false,
				}))
				Ω(typ).Should(Equal(resource.ResourceType("some-resource-type")))
				Ω(tags).Should(ConsistOf("some", "tags"))
			})

			It("gets the resource with the correct source, params, and version", func() {
				Ω(fakeResource.GetCallCount()).Should(Equal(1))

				_, gotSource, gotParams, gotVersion := fakeResource.GetArgsForCall(0)
				Ω(gotSource).Should(Equal(resourceConfig.Source))
				Ω(gotParams).Should(Equal(params))
				Ω(gotVersion).Should(Equal(version))
			})

			It("gets the resource with the io config forwarded", func() {
				Ω(fakeResource.GetCallCount()).Should(Equal(1))

func (fetcher Fetcher) FetchImage(
	logger lager.Logger,
	imageConfig atc.TaskImageConfig,
	signals <-chan os.Signal,
	identifier worker.Identifier,
	metadata worker.Metadata,
	delegate worker.ImageFetchingDelegate,
	worker worker.Client,
	customTypes atc.ResourceTypes,
) (worker.Image, error) {
	tracker := fetcher.trackerFactory.TrackerFor(worker)
	resourceType := resource.ResourceType(imageConfig.Type)

	checkSess := resource.Session{
		ID:       identifier,
		Metadata: metadata,
	}

	checkSess.ID.Stage = db.ContainerStageCheck
	checkSess.ID.ImageResourceType = imageConfig.Type
	checkSess.ID.ImageResourceSource = imageConfig.Source
	checkSess.Metadata.Type = db.ContainerTypeCheck
	checkSess.Metadata.WorkingDirectory = ""
	checkSess.Metadata.EnvironmentVariables = nil

	checkingResource, err := tracker.Init(
		logger.Session("check-image"),
		resource.EmptyMetadata{},
		checkSess,
		resourceType,
		nil,
		customTypes,
		delegate,
	)
	if err != nil {
		return nil, err
	}

	defer checkingResource.Release(nil)

	versions, err := checkingResource.Check(imageConfig.Source, nil)
	if err != nil {
		return nil, err
	}

	if len(versions) == 0 {
		return nil, ErrImageUnavailable
	}

	cacheID := resource.ResourceCacheIdentifier{
		Type:    resourceType,
		Version: versions[0],
		Source:  imageConfig.Source,
	}

	volumeID := cacheID.VolumeIdentifier()

	err = delegate.ImageVersionDetermined(volumeID)
	if err != nil {
		return nil, err
	}

	getSess := resource.Session{
		ID:       identifier,
		Metadata: metadata,
	}

	getSess.ID.Stage = db.ContainerStageGet
	getSess.ID.ImageResourceType = imageConfig.Type
	getSess.ID.ImageResourceSource = imageConfig.Source
	getSess.Metadata.Type = db.ContainerTypeGet
	getSess.Metadata.WorkingDirectory = ""
	getSess.Metadata.EnvironmentVariables = nil

	getResource, cache, err := tracker.InitWithCache(
		logger.Session("init-image"),
		resource.EmptyMetadata{},
		getSess,
		resourceType,
		nil,
		cacheID,
		customTypes,
		delegate,
	)
	if err != nil {
		return nil, err
	}

	isInitialized, err := cache.IsInitialized()
	if err != nil {
		return nil, err
	}

	versionedSource := getResource.Get(
		resource.IOConfig{
			Stderr: delegate.Stderr(),
		},
		imageConfig.Source,
		nil,
		versions[0],
//.........这里部分代码省略.........

func (scanner *resourceScanner) scan(
	logger lager.Logger,
	resourceConfig atc.ResourceConfig,
	resourceTypes atc.ResourceTypes,
	savedResource db.SavedResource,
	fromVersion atc.Version,
) error {
	pipelinePaused, err := scanner.db.IsPaused()
	if err != nil {
		logger.Error("failed-to-check-if-pipeline-paused", err)
		return err
	}

	if pipelinePaused {
		logger.Debug("pipeline-paused")
		return nil
	}

	if savedResource.Paused {
		logger.Debug("resource-paused")
		return nil
	}

	pipelineID := scanner.db.GetPipelineID()

	var resourceTypeVersion atc.Version
	_, found := resourceTypes.Lookup(resourceConfig.Type)
	if found {
		savedResourceType, resourceTypeFound, err := scanner.db.GetResourceType(resourceConfig.Type)
		if err != nil {
			logger.Error("failed-to-find-resource-type", err)
			return err
		}
		if resourceTypeFound {
			resourceTypeVersion = atc.Version(savedResourceType.Version)
		}
	}

	session := resource.Session{
		ID: worker.Identifier{
			ResourceTypeVersion: resourceTypeVersion,
			ResourceID:          savedResource.ID,
			Stage:               db.ContainerStageRun,
			CheckType:           resourceConfig.Type,
			CheckSource:         resourceConfig.Source,
		},
		Metadata: worker.Metadata{
			Type:       db.ContainerTypeCheck,
			PipelineID: pipelineID,
		},
		Ephemeral: true,
	}

	res, err := scanner.tracker.Init(
		logger,
		resource.TrackerMetadata{
			ResourceName: resourceConfig.Name,
			PipelineName: savedResource.PipelineName,
			ExternalURL:  scanner.externalURL,
		},
		session,
		resource.ResourceType(resourceConfig.Type),
		[]string{},
		resourceTypes,
		worker.NoopImageFetchingDelegate{},
	)
	if err != nil {
		logger.Error("failed-to-initialize-new-resource", err)
		return err
	}

	defer res.Release(nil)

	logger.Debug("checking", lager.Data{
		"from": fromVersion,
	})

	newVersions, err := res.Check(resourceConfig.Source, fromVersion)

	setErr := scanner.db.SetResourceCheckError(savedResource, err)
	if setErr != nil {
		logger.Error("failed-to-set-check-error", err)
	}

	if err != nil {
		if rErr, ok := err.(resource.ErrResourceScriptFailed); ok {
			logger.Info("check-failed", lager.Data{"exit-status": rErr.ExitStatus})
			return rErr
		}

		logger.Error("failed-to-check", err)
		return err
	}

	if len(newVersions) == 0 {
		logger.Debug("no-new-versions")
		return nil
	}

	logger.Info("versions-found", lager.Data{
//.........这里部分代码省略.........

// Run ultimately registers the configured resource version's ArtifactSource
// under the configured SourceName. How it actually does this is determined by
// a few factors.
//
// First, a worker that supports the given resource type is chosen, and a
// container is created on the worker.
//
// If the worker has a VolumeManager, and its cache is already warmed, the
// cache will be mounted into the container, and no fetching will be performed.
// The container will be used to stream the contents of the cache to later
// steps that require the artifact but are running on a worker that does not
// have the cache.
//
// If the worker does not have a VolumeManager, or if the worker does have a
// VolumeManager but a cache for the version of the resource is not present,
// the specified version of the resource will be fetched. As long as running
// the fetch script works, Run will return nil regardless of its exit status.
//
// If the worker has a VolumeManager but did not have the cache initially, the
// fetched ArtifactSource is initialized, thus warming the worker's cache.
//
// At the end, the resulting ArtifactSource (either from using the cache or
// fetching the resource) is registered under the step's SourceName.
func (step *GetStep) Run(signals <-chan os.Signal, ready chan<- struct{}) error {
	trackedResource, cache, err := step.tracker.InitWithCache(
		step.logger,
		step.stepMetadata,
		step.session,
		resource.ResourceType(step.resourceConfig.Type),
		step.tags,
		step.cacheIdentifier,
	)
	if err != nil {
		step.logger.Error("failed-to-initialize-resource", err)
		return err
	}

	step.resource = trackedResource

	step.versionedSource = step.resource.Get(
		resource.IOConfig{
			Stdout: step.delegate.Stdout(),
			Stderr: step.delegate.Stderr(),
		},
		step.resourceConfig.Source,
		step.params,
		step.version,
	)

	isInitialized, err := cache.IsInitialized()
	if err != nil {
		step.logger.Error("failed-to-check-if-cache-is-initialized", err)
		return err
	}

	if isInitialized {
		step.logger.Debug("cache-already-initialized")

		fmt.Fprintf(step.delegate.Stdout(), "using version of resource found in cache\n")
		close(ready)
	} else {
		step.logger.Debug("cache-not-initialized")

		err = step.versionedSource.Run(signals, ready)

		if err, ok := err.(resource.ErrResourceScriptFailed); ok {
			step.delegate.Completed(ExitStatus(err.ExitStatus), nil)
			return nil
		}

		if err == resource.ErrAborted {
			return ErrInterrupted
		}

		if err != nil {
			step.logger.Error("failed-to-run-get", err)
			return err
		}

		err = cache.Initialize()
		if err != nil {
			step.logger.Error("failed-to-initialize-cache", err)
		}
	}

	step.repository.RegisterSource(step.sourceName, step)

	step.succeeded = true
	step.delegate.Completed(ExitStatus(0), &VersionInfo{
		Version:  step.versionedSource.Version(),
		Metadata: step.versionedSource.Metadata(),
	})

	return nil
}

func (radar *Radar) scan(logger lager.Logger, resourceConfig atc.ResourceConfig, resourceTypes atc.ResourceTypes, savedResource db.SavedResource) error {
	pipelinePaused, err := radar.db.IsPaused()
	if err != nil {
		logger.Error("failed-to-check-if-pipeline-paused", err)
		return err
	}

	if pipelinePaused {
		logger.Debug("pipeline-paused")
		return nil
	}

	if savedResource.Paused {
		logger.Debug("resource-paused")
		return nil
	}

	session := resource.Session{
		ID: worker.Identifier{
			ResourceID:  savedResource.ID,
			Stage:       db.ContainerStageRun,
			CheckType:   resourceConfig.Type,
			CheckSource: resourceConfig.Source,
		},
		Metadata: worker.Metadata{
			Type:         db.ContainerTypeCheck,
			PipelineName: radar.db.GetPipelineName(),
		},
		Ephemeral: true,
	}

	res, err := radar.tracker.Init(
		logger,
		resource.EmptyMetadata{},
		session,
		resource.ResourceType(resourceConfig.Type),
		[]string{},
		resourceTypes,
		worker.NoopImageFetchingDelegate{},
	)
	if err != nil {
		logger.Error("failed-to-initialize-new-resource", err)
		return err
	}

	defer res.Release(nil)

	vr, found, err := radar.db.GetLatestVersionedResource(savedResource)
	if err != nil {
		logger.Error("failed-to-get-current-version", err)
		return err
	}

	var from db.Version
	if found {
		from = vr.Version
	}

	logger.Debug("checking", lager.Data{
		"from": from,
	})

	newVersions, err := res.Check(resourceConfig.Source, atc.Version(from))

	setErr := radar.db.SetResourceCheckError(savedResource, err)
	if setErr != nil {
		logger.Error("failed-to-set-check-error", err)
	}

	if err != nil {
		if rErr, ok := err.(resource.ErrResourceScriptFailed); ok {
			logger.Info("check-failed", lager.Data{"exit-status": rErr.ExitStatus})
			return nil
		}

		logger.Error("failed-to-check", err)
		return err
	}

	if len(newVersions) == 0 {
		logger.Debug("no-new-versions")
		return nil
	}

	logger.Info("versions-found", lager.Data{
		"versions": newVersions,
		"total":    len(newVersions),
	})

	err = radar.db.SaveResourceVersions(resourceConfig, newVersions)
	if err != nil {
		logger.Error("failed-to-save-versions", err, lager.Data{
			"versions": newVersions,
		})
	}

	return nil
}

// Run chooses a worker that supports the step's resource type and creates a
// container.
//
// All ArtifactSources present in the SourceRepository are then brought into
// the container, using volumes if possible, and streaming content over if not.
//
// The resource's put script is then invoked. The PutStep is ready as soon as
// the resource's script starts, and signals will be forwarded to the script.
func (step *PutStep) Run(signals <-chan os.Signal, ready chan<- struct{}) error {
	sources := step.repository.AsMap()

	resourceSources := make(map[string]resource.ArtifactSource)
	for name, source := range sources {
		resourceSources[string(name)] = resourceSource{source}
	}

	trackedResource, missingNames, err := step.tracker.InitWithSources(
		step.logger,
		step.stepMetadata,
		step.session,
		resource.ResourceType(step.resourceConfig.Type),
		step.tags,
		resourceSources,
	)

	if err != nil {
		return err
	}

	missingSourceNames := make([]SourceName, len(missingNames))
	for i, n := range missingNames {
		missingSourceNames[i] = SourceName(n)
	}

	step.resource = trackedResource

	scopedRepo, err := step.repository.ScopedTo(missingSourceNames...)
	if err != nil {
		return err
	}

	step.versionedSource = step.resource.Put(
		resource.IOConfig{
			Stdout: step.delegate.Stdout(),
			Stderr: step.delegate.Stderr(),
		},
		step.resourceConfig.Source,
		step.params,
		resourceSource{scopedRepo},
	)

	err = step.versionedSource.Run(signals, ready)

	if err, ok := err.(resource.ErrResourceScriptFailed); ok {
		step.delegate.Completed(ExitStatus(err.ExitStatus), nil)
		return nil
	}

	if err == resource.ErrAborted {
		return ErrInterrupted
	}

	if err != nil {
		return err
	}

	step.succeeded = true
	step.delegate.Completed(ExitStatus(0), &VersionInfo{
		Version:  step.versionedSource.Version(),
		Metadata: step.versionedSource.Metadata(),
	})

	return nil
}

			Eventually(times).Should(Receive())

			sessionID, typ, tags := fakeTracker.InitArgsForCall(0)
			Ω(sessionID).Should(Equal(resource.Session{
				ID: worker.Identifier{
					PipelineName: "some-pipeline-name",

					Name: "some-resource",
					Type: "check",

					CheckType:   "git",
					CheckSource: resourceConfig.Source,
				},
				Ephemeral: true,
			}))
			Ω(typ).Should(Equal(resource.ResourceType("git")))
			Ω(tags).Should(BeEmpty()) // This allows the check to run on any worker
		})

		It("checks on a specified interval", func() {
			var time1 time.Time
			var time2 time.Time

			Eventually(times).Should(Receive(&time1))
			Eventually(times).Should(Receive(&time2))

			Ω(time2.Sub(time1)).Should(BeNumerically("~", interval, interval/4))
		})

		It("grabs a resource checking lock before checking, releases after done", func() {
			Eventually(times).Should(Receive())

					Metadata: worker.Metadata{
						Type:         db.ContainerTypeCheck,
						PipelineName: "some-pipeline",
					},
					Ephemeral: true,
				}))
				Expect(customTypes).To(Equal(atc.ResourceTypes{
					{
						Name:   "some-custom-resource",
						Type:   "docker-image",
						Source: atc.Source{"custom": "source"},
					},
				}))
				Expect(delegate).To(Equal(worker.NoopImageFetchingDelegate{}))

				Expect(typ).To(Equal(resource.ResourceType("git")))
				Expect(tags).To(BeEmpty()) // This allows the check to run on any worker
			})

			Context("when the resource config has a specified check interval", func() {
				BeforeEach(func() {
					resourceConfig.CheckEvery = "10ms"

					fakeRadarDB.GetConfigReturns(atc.Config{
						Resources: atc.ResourceConfigs{
							resourceConfig,
						},
					}, 1, true, nil)
				})

				It("checks using the specified interval instead of the default", func() {

func (radar *Radar) scan(logger lager.Logger, resourceConfig atc.ResourceConfig, savedResource db.SavedResource) error {
	pipelinePaused, err := radar.db.IsPaused()
	if err != nil {
		logger.Error("failed-to-check-if-pipeline-paused", err)
		return err
	}

	if pipelinePaused {
		logger.Debug("pipeline-paused")
		return nil
	}

	if savedResource.Paused {
		logger.Debug("resource-paused")
		return nil
	}

	typ := resource.ResourceType(resourceConfig.Type)

	res, err := radar.tracker.Init(
		logger,
		resource.EmptyMetadata{},
		checkIdentifier(radar.db.GetPipelineName(), resourceConfig),
		typ,
		[]string{},
	)
	if err != nil {
		logger.Error("failed-to-initialize-new-resource", err)
		return err
	}

	defer res.Release(0)

	vr, found, err := radar.db.GetLatestVersionedResource(savedResource)
	if err != nil {
		logger.Error("failed-to-get-current-version", err)
		return err
	}

	var from db.Version
	if found {
		from = vr.Version
	}

	logger.Debug("checking", lager.Data{
		"from": from,
	})

	newVersions, err := res.Check(resourceConfig.Source, atc.Version(from))

	setErr := radar.db.SetResourceCheckError(savedResource, err)
	if setErr != nil {
		logger.Error("failed-to-set-check-error", err)
	}

	if err != nil {
		if rErr, ok := err.(resource.ErrResourceScriptFailed); ok {
			logger.Info("check-failed", lager.Data{"exit-status": rErr.ExitStatus})
			return nil
		}

		logger.Error("failed-to-check", err)
		return err
	}

	if len(newVersions) == 0 {
		logger.Debug("no-new-versions")
		return nil
	}

	logger.Info("versions-found", lager.Data{
		"versions": newVersions,
		"total":    len(newVersions),
	})

	err = radar.db.SaveResourceVersions(resourceConfig, newVersions)
	if err != nil {
		logger.Error("failed-to-save-versions", err, lager.Data{
			"versions": newVersions,
		})
	}

	return nil
}

func (radar *Radar) scan(logger lager.Logger, resourceName string) error {
	pipelinePaused, err := radar.db.IsPaused()
	if err != nil {
		logger.Error("failed-to-check-if-pipeline-paused", err)
		return err
	}

	if pipelinePaused {
		logger.Debug("pipeline-paused")
		return nil
	}

	config, _, err := radar.db.GetConfig()
	if err != nil {
		logger.Error("failed-to-get-config", err)
		// don't propagate error; we can just retry next tick
		return nil
	}

	resourceConfig, found := config.Resources.Lookup(resourceName)
	if !found {
		logger.Info("resource-removed-from-configuration")
		// return an error so that we exit
		return resourceNotConfiguredError{ResourceName: resourceName}
	}

	savedResource, err := radar.db.GetResource(resourceName)
	if err != nil {
		return err
	}

	if savedResource.Paused {
		return nil
	}

	typ := resource.ResourceType(resourceConfig.Type)

	res, err := radar.tracker.Init(checkIdentifier(radar.db.GetPipelineName(), resourceConfig), typ, []string{})
	if err != nil {
		logger.Error("failed-to-initialize-new-resource", err)
		return err
	}

	defer res.Release()

	var from db.Version
	if vr, err := radar.db.GetLatestVersionedResource(savedResource); err == nil {
		from = vr.Version
	}

	logger.Debug("checking", lager.Data{
		"from": from,
	})

	newVersions, err := res.Check(resourceConfig.Source, atc.Version(from))
	setErr := radar.db.SetResourceCheckError(savedResource, err)
	if setErr != nil {
		logger.Error("failed-to-set-check-error", err)
	}

	if err != nil {
		logger.Error("failed-to-check", err)

		return err
	}

	if len(newVersions) == 0 {
		logger.Debug("no-new-versions")
		return nil
	}

	logger.Info("versions-found", lager.Data{
		"versions": newVersions,
		"total":    len(newVersions),
	})

	err = radar.db.SaveResourceVersions(resourceConfig, newVersions)
	if err != nil {
		logger.Error("failed-to-save-versions", err, lager.Data{
			"versions": newVersions,
		})
	}

	return nil
}