Skip to content
This repository has been archived by the owner on Nov 25, 2020. It is now read-only.

Commit

Permalink
Fixes to upload and cache
Browse files Browse the repository at this point in the history
  • Loading branch information
@ghecquet
ghecquet committed Jun 30, 2016
1 parent 488ac7a commit 5d5ac4b
Show file tree
Hide file tree
Showing 11 changed files with 267 additions and 118 deletions.
6 changes: 2 additions & 4 deletions core/src/plugins/core.access/src/Stream/AuthStream.php
View file
Expand Up @@ -14,8 +14,7 @@
use Pydio\Auth\Core\MemorySafe;
use Pydio\Core\Model\ContextInterface;

class AuthStream implements StreamInterface
{
class AuthStream implements StreamInterface {
use StreamDecoratorTrait;

/** @var ContextInterface Context */
Expand Down Expand Up @@ -43,8 +42,7 @@ public function __construct(
"auth" => $auth
]);

$resource = PydioStreamWrapper::getResource($stream);
$this->stream = new Stream($resource, $node);
$this->stream = $stream;
}

public function getContents() {
Expand Down
19 changes: 7 additions & 12 deletions core/src/plugins/core.access/src/Stream/MetadataCachingStream.php
View file
Expand Up @@ -11,7 +11,6 @@
use GuzzleHttp\Stream\StreamInterface;
use Pydio\Access\Core\Model\AJXP_Node;
use Pydio\Cache\Core\AbstractCacheDriver;
use Pydio\Cache\Core\CacheStreamLayer;
use Pydio\Core\Services\CacheService;

/**
Expand All @@ -37,8 +36,8 @@ class MetadataCachingStream implements StreamInterface
/** @var string path */
private $path;

/** @var string statCacheId */
private $cacheId;
/** @var array statCacheId */
private $cacheOptions;

/**
* We will treat the buffer object as the body of the stream
Expand All @@ -54,7 +53,7 @@ public function __construct(
) {
$this->node = $node;
$this->uri = $node->getUrl();
$this->cacheId = AbstractCacheDriver::computeIdForNode($node, "meta");
$this->cacheOptions = AbstractCacheDriver::getOptionsForNode($node, "meta");
$this->contentFilters = $node->getRepository()->getContentFilter()->filters;
$this->path = parse_url($this->uri, PHP_URL_PATH);

Expand All @@ -63,10 +62,6 @@ public function __construct(
$this->stat();
}

public function __destruct() {
$this->stream->close();
}

public function getSize() {
$stat = $this->stat();

Expand Down Expand Up @@ -102,7 +97,7 @@ public function getContents() {

public function stat() {

$stat = CacheService::fetch(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheId);
$stat = CacheService::fetch(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheOptions["id"]);

if(is_array($stat)) return $stat;

Expand All @@ -120,14 +115,14 @@ public function stat() {

$node = new AJXP_Node($this->uri . "/" . $path);

$id = AbstractCacheDriver::computeIdForNode($node, "meta");
$id = AbstractCacheDriver::getOptionsForNode($node, "meta")["id"];

CacheService::save(AJXP_CACHE_SERVICE_NS_NODES, $id, $stat);
}
} else {
CacheService::save(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheId, $stats);
CacheService::save(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheOptions["id"], $stats, $this->cacheOptions["timelimit"]);
}

return CacheService::fetch(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheId);
return CacheService::fetch(AJXP_CACHE_SERVICE_NS_NODES, $this->cacheOptions["id"]);
}
}
130 changes: 88 additions & 42 deletions core/src/plugins/core.access/src/Stream/Stream.php
View file
Expand Up @@ -3,9 +3,10 @@

use Exception;
use Guzzle\Service\Loader\JsonLoader;
use GuzzleHttp\Client;
use GuzzleHttp\Command\Command;
use GuzzleHttp\Command\Guzzle\Description;
use GuzzleHttp\Command\Guzzle\GuzzleClient;
use GuzzleHttp\Client as HTTPClient;
use GuzzleHttp\Stream\GuzzleStreamWrapper;
use GuzzleHttp\Stream\Stream as GuzzleStream;
use GuzzleHttp\Stream\StreamInterface;
Expand All @@ -30,6 +31,12 @@ class Stream implements StreamInterface
/** @var GuzzleClient $client */
private $client;

/** @var Client $httpClient */
private $httpClient;

/** @var Command $command */
private $command;

private $seekable = true;
private $readable = true;
private $writable = true;
Expand Down Expand Up @@ -62,11 +69,20 @@ public function __construct(
AJXP_Node $node,
$options = []
) {

$this->node = $node;

$ctx = $node->getContext();
$repository = $ctx->getRepository();


$this->customMetadata["uri"] = $node->getUrl();

$this->attach($resource);

$this->readable = isset($options['readable']) ? $options['readable'] : $this->readable;
$this->writable = isset($options['writable']) ? $options['writable'] : $this->writable;

$apiUrl = $repository->getContextOption($ctx, "API_URL");
$host = $repository->getContextOption($ctx, "HOST");
$uri = $repository->getContextOption($ctx, "URI");
Expand All @@ -80,35 +96,35 @@ public function __construct(
}

$options["base_url"] = $apiUrl;
$this->httpClient = new Client([
"base_url" => $apiUrl
]);

$options["defaults"] = self::getContextOption($ctx);
$resources = $options["defaults"]["resources"];
$options["defaults"] = array_intersect_key($options["defaults"], ["subscribers" => "", "auth" => ""]);

// Creating Guzzle instances
$httpClient = new HTTPClient($options);
$locator = new FileLocator([dirname($resources)]);
$jsonLoader = new JsonLoader($locator);
$description = $jsonLoader->load($locator->locate(basename($resources)));
$description = new Description($description);
$client = new GuzzleClient($httpClient, $description, $options);
$client = new GuzzleClient($this->httpClient, $description);

foreach ($options["defaults"]["subscribers"] as $subscriber) {
$client->getEmitter()->attach($subscriber);
}

$stream = Stream::factory($resource);
$resource = PydioStreamWrapper::getResource($stream);
$this->attach($resource);

$this->node = $node;
$this->client = $client;
}

public static function factory($resource = '', array $options = [])
{
public static function factory($resource = '', $mode = "r+", array $options = []) {
if ($resource instanceof AJXP_Node) {
$stream = fopen('php://memory', 'r+');

return new self($stream, $resource, $options);
$node = $resource;

return new self(fopen('php://memory', $mode), $node, [
"readable" => isset(self::$readWriteHash['read'][$mode]),
"writable" => isset(self::$readWriteHash['write'][$mode])
]);
}

return GuzzleStream::factory($resource, $options);
Expand Down Expand Up @@ -192,8 +208,8 @@ public function detach()
public function attach($stream) {
$this->resource = $stream;
$meta = stream_get_meta_data($this->resource);

$this->seekable = $meta['seekable'];

$this->readable = isset(self::$readWriteHash['read'][$meta['mode']]);
$this->writable = isset(self::$readWriteHash['write'][$meta['mode']]);
}
Expand Down Expand Up @@ -304,6 +320,8 @@ public function seek($offset, $whence = SEEK_SET) {
/**
* Give a relative tell()
* {@inheritdoc}
*/
public function tell() {
return $this->resource ? ftell($this->resource) : false;
Expand All @@ -313,18 +331,29 @@ public function read($length) {
return $this->readable ? fread($this->resource, $length) : false;
}

public function write($string) {
public function write($buffer) {
// We can't know the size after writing anything
$this->size = null;

$this->detach();

$stream = Stream::factory($string);
$type = gettype($buffer);

if ($type == "string") {
$stream = Stream::factory($buffer);
} else {
$stream = $buffer;
$stream->seek(0);
}

$this->attach(GuzzleStreamWrapper::getResource($stream));

$this->size = $stream->getSize();

$command = $this->client->getCommand('Put', [
'path' => $this->node,
'body' => $stream
'path' => $this->node,
'headers' => [
"Content-Length" => $this->size
],
'body' => $this->resource
]);

$this->client->execute($command);
Expand All @@ -344,22 +373,42 @@ public function getMetadata($key = null) {
}
}

private function ls() {
$command = $this->client->getCommand('Ls', [
private function prepare($cmdName = null) {

$options = self::getContextOption($this->node->getContext());
$options = array_intersect_key($options, ["subscribers" => "", "auth" => ""]);

if (!isset($this->httpClient)) {
$this->httpClient = new Client($options);
} else {
foreach ($options as $key => $option) {
$this->httpClient->setDefaultOption($key, $option);
}
}

if (!isset($cmdName)) {
return;
}

$this->command = $this->client->getCommand($cmdName, [
'path' => $this->node
]);
}

$result = $this->client->execute($command);
private function ls() {

$this->prepare('Ls');

$result = $this->client->execute($this->command);

return $result;
}

private function get() {
$command = $this->client->getCommand('Get', [
'path' => $this->node
]);

$result = $this->client->execute($command);
$this->prepare('Get');

$result = $this->client->execute($this->command);

$this->detach();
$this->attach(GuzzleStreamWrapper::getResource($result["body"]));
Expand All @@ -368,12 +417,11 @@ private function get() {
}

public function stat() {
$command = $this->client->getCommand('Stat', [
'path' => $this->node
]);

$this->prepare('Stat');

try {
$result = $this->client->execute($command);
$result = $this->client->execute($this->command);
} catch (Exception $e) {
return null;
}
Expand All @@ -382,32 +430,30 @@ public function stat() {
}

public function mkdir() {
$command = $this->client->getCommand('Mkdir', [
'path' => $this->node
]);

$this->client->execute($command);
$this->prepare('Mkdir');

$this->client->execute($this->command);

return true;
}

public function rmdir() {
$command = $this->client->getCommand('Rmdir', [
'path' => $this->node
]);

$this->client->execute($command);
$this->prepare('Rmdir');

$this->client->execute($this->command);

return true;
}

public function rename($newNode) {
$command = $this->client->getCommand('Rename', [
$this->client->getCommand('Rename', [
'path' => $this->node,
'newPath' => $newNode
]);

$this->client->execute($command);
$this->client->execute($this->command);

return true;
}
Expand Down
22 changes: 18 additions & 4 deletions core/src/plugins/core.access/src/Stream/StreamWrapper.php
View file
Expand Up @@ -59,7 +59,11 @@ public static function register($protocol) {
}

public function stream_open($path, $mode, $options, &$opened_path) {
$this->stream = self::createStream($path);
$this->stream = self::createStream($path, $mode);

if ($this->stream->isWritable() && !$this->stream->isReadable()) {
$a = 1;
}

return true;
}
Expand Down Expand Up @@ -88,6 +92,10 @@ public function stream_seek($offset, $whence) {
return $this->stream->seek($offset, $whence);
}

public function stream_close() {
$this->stream->close();
}

/**
* @param $path
* @param $options
Expand Down Expand Up @@ -219,20 +227,26 @@ public static function copyStreamInStream($from, $to) {
}
}

public static function createStream($path) {
public static function changeMode($path, $chmodValue) {
}

public static function createStream($path, $mode = "r+") {
$node = new AJXP_Node($path);
$repository = $node->getRepository();
$ctx = $node->getContext();

$useOAuthStream = $repository->getContextOption($ctx, "USE_OAUTH_STREAM", false);
$useAuthStream = $repository->getContextOption($ctx, "USE_AUTH_STREAM", !$useOAuthStream);

$nodeStream = Stream::factory($node);
$nodeStream = Stream::factory($node, $mode);
if ($useAuthStream) $nodeStream = new AuthStream($nodeStream, $node);
if ($useOAuthStream) $nodeStream = new OAuthStream($nodeStream, $node);

$nodeStream = new MetadataCachingStream($nodeStream, $node);
$nodeStream = new WriteBufferStream($nodeStream, $node);

return $nodeStream;
PydioStreamWrapper::getResource($nodeStream);

return $nodeStream;
}
}

0 comments on commit 5d5ac4b

Please sign in to comment.