Viewing file:  BatchRunner.php (6.17 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
<?php
/**
* Copyright 2017 Google Inc. All Rights Reserved.
*
* 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.
*/
namespace Google\Cloud\Core\Batch;
use Google\Cloud\Core\SysvTrait;
/**
* A class for executing jobs in batch.
*
* @experimental The experimental flag means that while we believe this method
* or class is ready for use, it may change before release in backwards-
* incompatible ways. Please use with caution, and test thoroughly when
* upgrading.
*/
class BatchRunner
{
use BatchDaemonTrait;
use SysvTrait;
/**
* @var JobConfig
*/
private $config;
/**
* @var ConfigStorageInterface
*/
private $configStorage;
/**
* @var ProcessItemInterface
*/
private $processor;
/**
* Determine internal implementation and loads the configuration.
*
* @param ConfigStorageInterface $configStorage [optional] The
* ConfigStorage object to use. **Defaults to** null. This is only
* for testing purpose.
* @param ProcessItemInterface $processor [optional] The processor object
* to use. **Defaults to** null. This is only for testing purpose.
*/
public function __construct(
ConfigStorageInterface $configStorage = null,
ProcessItemInterface $processor = null
) {
if ($configStorage === null || $processor === null) {
if ($this->isSysvIPCLoaded() && $this->isDaemonRunning()) {
$configStorage = new SysvConfigStorage();
$processor = new SysvProcessor();
} else {
$configStorage = InMemoryConfigStorage::getInstance();
$processor = $configStorage;
}
}
$this->configStorage = $configStorage;
$this->processor = $processor;
$this->loadConfig();
}
/**
* Register a job for batch execution.
*
* @param string $identifier Unique identifier of the job.
* @param callable $func Any Callable except for Closure. The callable
* should accept an array of items as the first argument.
* @param array $options [optional] {
* Configuration options.
*
* @type int $batchSize The size of the batch.
* @type float $callPeriod The period in seconds from the last execution
* to force executing the job.
* @type int $numWorkers The number of child processes. It only takes
* effect with the {@see \Google\Cloud\Core\Batch\BatchDaemon}.
* @type string $bootstrapFile A file to load before executing the
* job. It's needed for registering global functions.
* }
* @return bool true on success, false on failure
* @throws \InvalidArgumentException When receiving a Closure.
*/
public function registerJob($identifier, $func, array $options = [])
{
if ($func instanceof \Closure) {
throw new \InvalidArgumentException('Closure is not allowed');
}
// Always work on the latest data
$result = $this->configStorage->lock();
if ($result === false) {
return false;
}
$this->config = $this->configStorage->load();
$this->config->registerJob(
$identifier,
function ($id) use ($identifier, $func, $options) {
return new BatchJob($identifier, $func, $id, $options);
}
);
try {
$result = $this->configStorage->save($this->config);
} finally {
$this->configStorage->unlock();
}
return $result;
}
/**
* Submit an item.
*
* @param string $identifier Unique identifier of the job.
* @param mixed $item It needs to be serializable.
*
* @return void
* @throws \RuntimeException
*/
public function submitItem($identifier, $item)
{
$job = $this->getJobFromId($identifier);
if ($job === null) {
throw new \RuntimeException(
"The identifier does not exist: $identifier"
);
}
$idNum = $job->id();
$this->processor->submit($item, $idNum);
}
/**
* Get the job with the given identifier.
*
* @param string $identifier Unique identifier of the job.
*
* @return BatchJob|null
*/
public function getJobFromId($identifier)
{
return $this->config->getJobFromId($identifier);
}
/**
* Get the job with the given numeric id.
*
* @param int $idNum A numeric id of the job.
*
* @return BatchJob|null
*/
public function getJobFromIdNum($idNum)
{
return $this->config->getJobFromIdNum($idNum);
}
/**
* Get all the jobs.
*
* @return BatchJob[]
*/
public function getJobs()
{
return $this->config->getJobs();
}
/**
* Load the config from the storage.
*
* @return bool true on success
* @throws \RuntimeException when it fails to load the config.
*/
public function loadConfig()
{
$result = $this->configStorage->lock();
if ($result === false) {
throw new \RuntimeException('Failed to lock the configStorage');
}
try {
$result = $this->configStorage->load();
} catch (\RuntimeException $e) {
$this->configStorage->clear();
throw $e;
} finally {
$this->configStorage->unlock();
}
$this->config = $result;
return true;
}
/**
* Gets the item processor.
*
* @return ProcessItemInterface
*/
public function getProcessor()
{
return $this->processor;
}
}
|