Chromium Code Reviews| Index: example/src/examples/cow_repository.dart |
| diff --git a/example/src/examples/cow_repository.dart b/example/src/examples/cow_repository.dart |
| new file mode 100644 |
| index 0000000000000000000000000000000000000000..49c486346ac021fa97f8542c3d1d94310ae8bc23 |
| --- /dev/null |
| +++ b/example/src/examples/cow_repository.dart |
| @@ -0,0 +1,136 @@ |
| +// Copyright (c) 2014, the Dart project authors. Please see the AUTHORS file |
| +// for details. All rights reserved. Use of this source code is governed by a |
| +// BSD-style license that can be found in the LICENSE file. |
| + |
| +library pub_server.copy_and_write_repository; |
| + |
| +import 'dart:async'; |
| + |
| +import 'package:logging/logging.dart'; |
| +import 'package:pub_server/repository.dart'; |
| + |
| +final Logger _logger = new Logger('pub_server.cow_repository'); |
| + |
| +/// A [CopyAndWriteRepository] writes to one repository and directs |
| +/// read-misses to another repository. |
| +/// |
| +/// Package versions not available from the read-write repository will be |
| +/// fetched from a read-fallback repository and uploaded to the read-write |
| +/// repository. This caches effectively all packages requested through this |
|
Søren Gjesse
2015/02/16 11:54:51
caches effectively -> effectively caches
kustermann
2015/02/16 14:44:02
Done.
|
| +/// [CopyAndWriteRepository]. |
| +/// |
| +/// New package versions which get uploaded will be stored only locally. |
| +class CopyAndWriteRepository extends PackageRepository { |
| + final PackageRepository local; |
| + final PackageRepository remote; |
| + final _RemoteMetadataCache _localCache; |
| + final _RemoteMetadataCache _remoteCache; |
| + |
| + /// Construct a new proxy with [local] as the local [PackageRepository] which |
| + /// is used for uploading new package versions to and [remote] as the |
| + /// read-only [PackageRepository] which is consulted on misses in [local]. |
| + CopyAndWriteRepository(PackageRepository local, PackageRepository remote) |
| + : this.local = local, |
| + this.remote = remote, |
| + this._localCache = new _RemoteMetadataCache(local), |
| + this._remoteCache = new _RemoteMetadataCache(remote); |
| + |
| + Stream<PackageVersion> versions(String package) { |
| + var controller; |
| + |
| + onListen() { |
| + Future.wait([_localCache.fetchVersionlist(package), |
| + _remoteCache.fetchVersionlist(package)]).then((tuple) { |
| + var versions = new Set() |
| + ..addAll(tuple[0]) |
| + ..addAll(tuple[1]); |
| + for (var version in versions) controller.add(version); |
| + controller.close(); |
| + }); |
| + } |
| + |
| + controller = new StreamController(onListen: onListen); |
| + return controller.stream; |
| + } |
| + |
| + Future<PackageVersion> lookupVersion(String package, String version) { |
| + return versions(package) |
| + .where((pv) => pv.versionString == version) |
| + .toList().then((List<PackageVersion> versions) { |
| + if (versions.length >= 1) return versions.first; |
| + return null; |
| + }); |
| + } |
| + |
| + Future<Stream> download(String package, String version) async { |
| + var packageVersion = await local.lookupVersion(package, version); |
| + |
| + if (packageVersion != null) { |
| + _logger.info('Serving $package/$version from local repository.'); |
| + return local.download(package, packageVersion.versionString); |
| + } else { |
| + // We first download the package from the remote repository and store |
| + // it locally. Then we read the local version and return it. |
| + |
| + _logger.info('Downloading $package/$version from remote repository.'); |
| + var stream = await remote.download(package, version); |
| + |
| + _logger.info('Upload $package/$version to local repository.'); |
| + await local.upload(stream); |
| + |
| + _logger.info('Serving $package/$version from local repository.'); |
| + return local.download(package, version); |
| + } |
| + } |
| + |
| + bool get supportsUpload => true; |
| + |
| + Future upload(Stream<List<int>> data) async { |
| + _logger.info('Starting upload to local package repository.'); |
| + var stream = await local.upload(data); |
| + |
| + // TODO: It's not really necessary to invalidate all. |
| + _logger.info('Upload finished. Invalidating in-memory cache.'); |
| + _localCache.invalidateAll(); |
| + } |
| + |
| + bool get supportsAsyncUpload => false; |
| +} |
| + |
| +/// A cache for [PackageVersion] objects for a given `package`. |
| +/// |
| +/// The constructor takes a [PackageRepository] which will be used to populate |
| +/// the cache. |
| +class _RemoteMetadataCache { |
| + final PackageRepository remote; |
| + |
| + Map<String, Set<PackageVersion>> _versions = {}; |
| + Map<String, Completer<Set<PackageVersion>>> _versionCompleters = {}; |
| + |
| + _RemoteMetadataCache(this.remote); |
| + |
| + // TODO: After a cache expiration we should invalidate entries and re-fetch |
| + // them. |
| + Future<List<PackageVersion>> fetchVersionlist(String package) { |
| + return _versionCompleters.putIfAbsent(package, () { |
| + var c = new Completer(); |
| + |
| + _versions.putIfAbsent(package, () => new Set()); |
| + remote.versions(package).toList().then((versions) { |
| + _versions[package].addAll(versions); |
| + c.complete(_versions[package]); |
| + }); |
| + |
| + return c; |
| + }).future.then((set) => set.toList()); |
| + } |
| + |
| + void addVersion(String package, PackageVersion version) { |
| + _versions.putIfAbsent(version.packageName, () => new Set()).add(version); |
| + } |
| + |
| + void invalidateAll() { |
| + _versionCompleters.clear(); |
| + _versions.clear(); |
| + } |
| +} |