Skip to content

Commit

Permalink
Merge c914f8c into bd57d79
Browse files Browse the repository at this point in the history
  • Loading branch information
@mcg-web
mcg-web committed Nov 10, 2016
2 parents bd57d79 + c914f8c commit f1f4ada
Show file tree
Hide file tree
Showing 8 changed files with 559 additions and 105 deletions.
6 changes: 5 additions & 1 deletion .travis.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,8 @@ before_install:

install: composer update --prefer-dist --no-interaction

script: if [ "$TRAVIS_PHP_VERSION" == "5.6" ]; then phpunit -d xdebug.max_nesting_level=1000 --debug --coverage-clover=coverage.clover; else phpunit --debug; fi
script: if [ "$TRAVIS_PHP_VERSION" == "5.6" ]; then phpunit -d xdebug.max_nesting_level=1000 --debug --coverage-clover build/logs/clover.xml; else phpunit --debug; fi

after_success:
- composer require "satooshi/php-coveralls:^1.0"
- travis_retry php vendor/bin/coveralls -v
5 changes: 5 additions & 0 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,3 +5,8 @@ fetching layer to provide a simplified and consistent API over various remote
data sources such as databases or web services via batching and caching.

This package is a PHP port of the [JS version](https://github.com/facebook/dataloader).

[![Build Status](https://travis-ci.org/overblog/dataloader-php.svg?branch=master)](https://travis-ci.org/overblog/dataloader-php)
[![Coverage Status](https://coveralls.io/repos/github/overblog/dataloader-php/badge.svg?branch=master)](https://coveralls.io/github/overblog/dataloader-php?branch=master)
[![Latest Stable Version](https://poser.pugx.org/overblog/dataloader-php/version)](https://packagist.org/packages/overblog/dataloader-php)
[![License](https://poser.pugx.org/overblog/dataloader-php/license)](https://packagist.org/packages/overblog/dataloader-php)
3 changes: 1 addition & 2 deletions composer.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,7 @@
},
"require": {
"php": "^5.5|^7.0",
"react/promise": "^2.4",
"symfony/cache": "^3.1"
"react/promise": "^2.4"
},
"require-dev": {
"phpunit/phpunit": "^4.1|^5.1"
Expand Down
4 changes: 2 additions & 2 deletions src/BatchLoadFn.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@ public function setBatchLoadFn(callable $batchLoadFn)
public function __invoke(array $keys)
{
$batchLoadFn = $this->getBatchLoadFn();
if (!is_callable($batchLoadFn)) {
throw new \RuntimeException('A batchLoadFn should be define.');
if (null === $batchLoadFn) {
throw new \LogicException('A valid batchLoadFn should be define.');
}

return $batchLoadFn($keys);
Expand Down
135 changes: 103 additions & 32 deletions src/DataLoader.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,18 @@ class DataLoader
*/
private $resolvedPromise;

/**
* @var self[]
*/
private static $instances = [];

public function __construct(BatchLoadFn $batchLoadFn, Option $options = null)
{
$this->batchLoadFn = $batchLoadFn;
$this->options = $options ?: new Option();
$this->eventLoop = class_exists('React\\EventLoop\\Factory') ? \React\EventLoop\Factory::create() : null;
$this->promiseCache = $this->options->getCacheMap();
self::$instances[] = $this;
}

/**
Expand All @@ -66,8 +72,7 @@ public function load($key)
// Determine options
$shouldBatch = $this->options->shouldBatch();
$shouldCache = $this->options->shouldCache();
$cacheKeyFn = $this->options->getCacheKeyFn();
$cacheKey = $cacheKeyFn ? $cacheKeyFn($key) : $key;
$cacheKey = $this->getCacheKeyFromKey($key);

// If caching and there is a cache-hit, return cached Promise.
if ($shouldCache) {
Expand All @@ -76,30 +81,38 @@ public function load($key)
return $cachedPromise;
}
}
$promise = null;

// Otherwise, produce a new Promise for this value.
$promise = new Promise(function ($resolve, $reject) use ($key, $shouldBatch) {
$this->queue[] = [
'key' => $key,
'resolve' => $resolve,
'reject' => $reject,
];

// Determine if a dispatch of this queue should be scheduled.
// A single dispatch should be scheduled per queue at the time when the
// queue changes from "empty" to "full".
if (count($this->queue) === 1) {
if ($shouldBatch) {
// If batching, schedule a task to dispatch the queue.
$this->enqueuePostPromiseJob(function () {
$promise = new Promise(
function ($resolve, $reject) use (&$promise, $key, $shouldBatch) {
$this->queue[] = [
'key' => $key,
'resolve' => $resolve,
'reject' => $reject,
'promise' => &$promise,
];

// Determine if a dispatch of this queue should be scheduled.
// A single dispatch should be scheduled per queue at the time when the
// queue changes from "empty" to "full".
if (count($this->queue) === 1) {
if ($shouldBatch) {
// If batching, schedule a task to dispatch the queue.
$this->enqueuePostPromiseJob(function () {
$this->dispatchQueue();
});
} else {
// Otherwise dispatch the (queue of one) immediately.
$this->dispatchQueue();
});
} else {
// Otherwise dispatch the (queue of one) immediately.
$this->dispatchQueue();
}
}
}
});
},
function (callable $resolve, callable $reject) {
// Cancel/abort any running operations like network connections, streams etc.

$reject(new \RuntimeException('DataLoader destroyed before promise complete.'));
});
// If caching, cache this promise.
if ($shouldCache) {
$this->promiseCache->set($cacheKey, $promise);
Expand Down Expand Up @@ -145,8 +158,8 @@ function ($key) {
public function clear($key)
{
$this->checkKey($key, __METHOD__);

$this->promiseCache->clear($key);
$cacheKey = $this->getCacheKeyFromKey($key);
$this->promiseCache->clear($cacheKey);

return $this;
}
Expand All @@ -160,6 +173,7 @@ public function clear($key)
public function clearAll()
{
$this->promiseCache->clearAll();

return $this;
}

Expand All @@ -174,8 +188,7 @@ public function prime($key, $value)
{
$this->checkKey($key, __METHOD__);

$cacheKeyFn = $this->options->getCacheKeyFn();
$cacheKey = $cacheKeyFn ? $cacheKeyFn($key) : $key;
$cacheKey = $this->getCacheKeyFromKey($key);

// Only add the key if it does not already exist.
if (!$this->promiseCache->has($cacheKey)) {
Expand All @@ -189,9 +202,34 @@ public function prime($key, $value)
return $this;
}

public function __destruct()
{
if ($this->needProcess()) {
foreach ($this->queue as $data) {
try {
$data['promise']->cancel();
} catch (\Exception $e) {
// no need to do nothing if cancel failed
}
}
}
foreach (self::$instances as $i => $instance) {
if ($this !== $instance) {
continue;
}
unset(self::$instances[$i]);
}
unset($this);
}

public function needProcess()
{
return count($this->queue) > 0;
}

public function process()
{
if (count($this->queue) > 0) {
if ($this->needProcess()) {
$this->dispatchQueue();
}
}
Expand All @@ -200,23 +238,22 @@ public function process()
* @param $promise
* @param bool $unwrap controls whether or not the value of the promise is returned for a fulfilled promise or if an exception is thrown if the promise is rejected
* @return mixed
* @throws null
*/
public function await($promise, $unwrap = true)
public static function await($promise, $unwrap = true)
{
$resolvedValue = null;
$exception = null;

if (!is_callable([$promise, 'then'])) {
throw new \InvalidArgumentException('Promise must have a "then" method.');
}
self::awaitInstances();

$promise->then(function ($values) use (&$resolvedValue) {
$resolvedValue = $values;
}, function ($reason) use (&$exception) {
$exception = $reason;
});
$this->process();
if ($exception instanceof \Exception) {
if (!$unwrap) {
return $exception;
Expand All @@ -227,6 +264,40 @@ public function await($promise, $unwrap = true)
return $resolvedValue;
}

private static function awaitInstances()
{
$dataLoaders = self::$instances;
if (empty($dataLoaders)) {
return;
}

$wait = true;

while ($wait) {
foreach ($dataLoaders as $dataLoader) {
try {
if (!$dataLoader || !$dataLoader->needProcess()) {
$wait = false;
continue;
}
$wait = true;
$dataLoader->process();
} catch (\Exception $e) {
$wait = false;
continue;
}
}
}
}

private function getCacheKeyFromKey($key)
{
$cacheKeyFn = $this->options->getCacheKeyFn();
$cacheKey = $cacheKeyFn ? $cacheKeyFn($key) : $key;

return $cacheKey;
}

private function checkKey($key, $method)
{
if (null === $key) {
Expand Down Expand Up @@ -290,8 +361,8 @@ private function dispatchQueueBatch(array $queue)
// Assert the expected response from batchLoadFn
if (!$batchPromise || !is_callable([$batchPromise, 'then'])) {
$this->failedDispatch($queue, new \RuntimeException(
'DataLoader must be constructed with a function which accepts '.
'Array<key> and returns Promise<Array<value>>, but the function did '.
'DataLoader must be constructed with a function which accepts ' .
'Array<key> and returns Promise<Array<value>>, but the function did ' .
sprintf('not return a Promise: %s.', gettype($batchPromise))
));

Expand Down
28 changes: 28 additions & 0 deletions tests/BatchLoadFnTest.php
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
<?php

/*
* This file is part of the DataLoaderPhp package.
*
* (c) Overblog <http://github.com/overblog/>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/

namespace Overblog\DataLoader\Tests;

use Overblog\DataLoader\BatchLoadFn;

class BatchLoadFnTest extends \PHPUnit_Framework_TestCase
{
/**
* @expectedException \LogicException
* @expectedExceptionMessage A valid batchLoadFn should be define.
*/
public function testNoBatchLoadFunctionGiven()
{
$batchLoadFn = new BatchLoadFn();

$batchLoadFn([]);
}
}

0 comments on commit f1f4ada

Please sign in to comment.