Add INotifyHandler to make notify more flexible

src/Change.php

@@ -0,0 +1,40 @@
+<?php
+/**
+ * @copyright Copyright (c) 2016 Robin Appelman <robin@icewind.nl>
+ * This file is licensed under the Licensed under the MIT license:
+ * http://opensource.org/licenses/MIT
+ *
+ */
+
+namespace Icewind\SMB;
+
+class Change {
+	private $code;
+
+	private $path;
+
+	/**
+	 * Change constructor.
+	 *
+	 * @param $code
+	 * @param $path
+	 */
+	public function __construct($code, $path) {
+		$this->code = $code;
+		$this->path = $path;
+	}
+
+	/**
+	 * @return integer
+	 */
+	public function getCode() {
+		return $this->code;
+	}
+
+	/**
+	 * @return string
+	 */
+	public function getPath() {
+		return $this->path;
+	}
+}

src/Connection.php

@@ -38,7 +38,7 @@ class Connection extends RawConnection {
 	 * get all unprocessed output from smbclient until the next prompt
 	 *
 	 * @param callable $callback (optional) callback to call for every line read
-	 * @return string
+	 * @return string[]
 	 * @throws AuthenticationException
 	 * @throws ConnectException
 	 * @throws ConnectionException
@@ -62,6 +62,7 @@ class Connection extends RawConnection {
 				$result = $callback($line);
 				if ($result === false) { // allow the callback to close the connection for infinite running commands
 					$this->close(true);
+					break;
 				}
 			} else {
 				$output[] .= $line;

src/INotifyHandler.php

@@ -0,0 +1,36 @@
+<?php
+/**
+ * @copyright Copyright (c) 2016 Robin Appelman <robin@icewind.nl>
+ * This file is licensed under the Licensed under the MIT license:
+ * http://opensource.org/licenses/MIT
+ *
+ */
+
+namespace Icewind\SMB;
+
+
+interface INotifyHandler {
+
+	/**
+	 * Get all changes detected since the start of the notify process or the last call to getChanges
+	 *
+	 * @return Change[]
+	 */
+	public function getChanges();
+
+	/**
+	 * Listen actively to all incoming changes
+	 *
+	 * Note that this is a blocking process and will cause the process to block forever if not explicitly terminated
+	 *
+	 * @param callable $callback
+	 */
+	public function listen($callback);
+
+	/**
+	 * Stop listening for changes
+	 *
+	 * Note that any pending changes will be discarded
+	 */
+	public function stop();
+}

src/IShare.php

@@ -145,8 +145,7 @@ interface IShare {
 
 	/**
 	 * @param string $path
-	 * @param callable $callback callable which will be called for each received change
+	 * @return INotifyHandler
-	 * @return mixed
 	 */
-	public function notify($path, callable $callback);
+	public function notify($path);
 }

src/NativeShare.php

@@ -290,14 +290,13 @@ class NativeShare extends AbstractShare {
 
 	/**
 	 * @param string $path
-	 * @param callable $callback callable which will be called for each received change
+	 * @return INotifyHandler
-	 * @return mixed
 	 */
-	public function notify($path, callable $callback) {
+	public function notify($path) {
 		// php-smbclient does support notify (https://github.com/eduardok/libsmbclient-php/issues/29)
 		// so we use the smbclient based backend for this
 		$share = new Share($this->server, $this->getName());
-		$share->notify($path, $callback);
+		return $share->notify($path);
 	}
 
 	public function __destruct() {

src/NotifyHandler.php

@@ -0,0 +1,88 @@
+<?php
+/**
+ * @copyright Copyright (c) 2016 Robin Appelman <robin@icewind.nl>
+ * This file is licensed under the Licensed under the MIT license:
+ * http://opensource.org/licenses/MIT
+ *
+ */
+
+namespace Icewind\SMB;
+
+
+class NotifyHandler implements INotifyHandler {
+	/**
+	 * @var Connection
+	 */
+	private $connection;
+
+	/**
+	 * @var string
+	 */
+	private $path;
+
+	private $listening = true;
+
+	/**
+	 * @param Connection $connection
+	 * @param string $path
+	 */
+	public function __construct(Connection $connection, $path) {
+		$this->connection = $connection;
+		$this->path = $path;
+	}
+
+	/**
+	 * Get all changes detected since the start of the notify process or the last call to getChanges
+	 *
+	 * @return Change[]
+	 */
+	public function getChanges() {
+		if (!$this->listening) {
+			return [];
+		}
+		stream_set_blocking($this->connection->getOutputStream(), 0);
+		$lines = [];
+		while (($line = $this->connection->readLine())) {
+			$lines[] = $line;
+		}
+		stream_set_blocking($this->connection->getOutputStream(), 1);
+		return array_values(array_filter(array_map([$this, 'parseChangeLine'], $lines)));
+	}
+
+	/**
+	 * Listen actively to all incoming changes
+	 *
+	 * Note that this is a blocking process and will cause the process to block forever if not explicitly terminated
+	 *
+	 * @param callable $callback
+	 */
+	public function listen($callback) {
+		if ($this->listening) {
+			$this->connection->read(function ($line) use ($callback) {
+				return $callback($this->parseChangeLine($line));
+			});
+		}
+	}
+
+	private function parseChangeLine($line) {
+		$code = (int)substr($line, 0, 4);
+		if ($code === 0) {
+			return null;
+		}
+		$subPath = str_replace('\\', '/', substr($line, 5));
+		if ($this->path === '') {
+			return new Change($code, $subPath);
+		} else {
+			return new Change($code, $this->path . '/' . $subPath);
+		}
+	}
+
+	public function stop() {
+		$this->listening = false;
+		$this->connection->close();
+	}
+
+	public function __destruct() {
+		$this->stop();
+	}
+}

src/Share.php

@@ -342,27 +342,18 @@ class Share extends AbstractShare {
 
 	/**
 	 * @param string $path
-	 * @param callable $callback callable which will be called for each received change
+	 * @return INotifyHandler
-	 * @return mixed
 	 * @throws ConnectionException
 	 * @throws DependencyException
 	 */
-	public function notify($path, callable $callback) {
+	public function notify($path) {
 		if (!$this->system->hasStdBuf()) { //stdbuf is required to disable smbclient's output buffering
 			throw new DependencyException('stdbuf is required for usage of the notify command');
 		}
 		$connection = $this->getConnection(); // use a fresh connection since the notify command blocks the process
 		$command = 'notify ' . $this->escapePath($path);
 		$connection->write($command . PHP_EOL);
-		$connection->read(function ($line) use ($callback, $path) {
+		return new NotifyHandler($connection, $path);
-			$code = (int)substr($line, 0, 4);
-			$subPath = str_replace('\\', '/', substr($line, 5));
-			if ($path === '') {
-				return $callback($code, $subPath);
-			} else {
-				return $callback($code, $path . '/' . $subPath);
-			}
-		});
 	}
 
 	/**

tests/NotifyHandlerTest.php

@@ -0,0 +1,103 @@
+<?php
+/**
+ * Copyright (c) 2016 Robin Appelman <icewind@owncloud.com>
+ * This file is licensed under the Licensed under the MIT license:
+ * http://opensource.org/licenses/MIT
+ */
+
+namespace Icewind\SMB\Test;
+
+use Icewind\SMB\Change;
+use Icewind\SMB\Exception\AlreadyExistsException;
+use Icewind\SMB\IShare;
+
+class NotifyHandlerTest extends TestCase {
+	/**
+	 * @var \Icewind\SMB\Server $server
+	 */
+	private $server;
+
+	private $config;
+
+	public function setUp() {
+		$this->requireBackendEnv('smbclient');
+		$this->config = json_decode(file_get_contents(__DIR__ . '/config.json'));
+		$this->server = new \Icewind\SMB\Server($this->config->host, $this->config->user, $this->config->password);
+	}
+
+	public function testGetChanges() {
+		$share = $this->server->getShare($this->config->share);
+		$process = $share->notify('');
+
+		$share->put(__FILE__, 'source.txt');
+		$share->rename('source.txt', 'target.txt');
+		$share->del('target.txt');
+		usleep(1000 * 100);// give it some time
+
+		$changes = $process->getChanges();
+		$process->stop();
+		$this->assertCount(5, $changes);
+		$this->assertEquals(IShare::NOTIFY_ADDED, $changes[0]->getCode());
+		$this->assertEquals('source.txt', $changes[0]->getPath());
+
+		$this->assertEquals(IShare::NOTIFY_RENAMED_OLD, $changes[1]->getCode());
+		$this->assertEquals('source.txt', $changes[1]->getPath());
+
+		$this->assertEquals(IShare::NOTIFY_RENAMED_NEW, $changes[2]->getCode());
+		$this->assertEquals('target.txt', $changes[2]->getPath());
+
+		$this->assertEquals(IShare::NOTIFY_MODIFIED, $changes[3]->getCode());
+		$this->assertEquals('target.txt', $changes[3]->getPath());
+
+		$this->assertEquals(IShare::NOTIFY_REMOVED, $changes[4]->getCode());
+		$this->assertEquals('target.txt', $changes[4]->getPath());
+	}
+
+	public function testChangesSubdir() {
+		$share = $this->server->getShare($this->config->share);
+
+		try {
+			$share->mkdir('sub');
+		} catch (AlreadyExistsException $e){
+
+		}
+		$process = $share->notify('sub');
+		usleep(1000 * 100);// give it some time to start listening
+		$share->put(__FILE__, 'sub/source.txt');
+		$share->del('sub/source.txt');
+		usleep(1000 * 100);// give it some time
+
+		$changes = $process->getChanges();
+		$process->stop();
+		$share->rmdir('sub');
+		$this->assertCount(2, $changes);
+		$this->assertEquals(IShare::NOTIFY_ADDED, $changes[0]->getCode());
+		$this->assertEquals('sub/source.txt', $changes[0]->getPath());
+	}
+
+	public function testListen() {
+		$share = $this->server->getShare($this->config->share);
+		$process = $share->notify('');
+
+		usleep(1000 * 100);// give it some time to start listening
+
+		$share->put(__FILE__, 'source.txt');
+		$share->del('source.txt');
+
+		$results = [];
+
+		// the notify process buffers incoming messages so callback will be triggered for the above changes
+		$process->listen(function ($change) use (&$results) {
+			$results = $change;
+			return false; // stop listening
+		});
+		$this->assertEquals($results, new Change(IShare::NOTIFY_ADDED, 'source.txt'));
+	}
+
+	public function testStopped() {
+		$share = $this->server->getShare($this->config->share);
+		$process = $share->notify('');
+		$process->stop();
+		$this->assertEquals([], $process->getChanges());
+	}
+}