Initial Commit

davidwdan · davidwdan · commit 68637bba80a3 · 2016-01-22T20:46:54.000-05:00
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,2 @@
+/composer.lock
+/vendor
diff --git a/.travis.yml b/.travis.yml
@@ -0,0 +1,17 @@
+language: php
+sudo: required
+
+php:
+  - 5.6
+  - 7
+  - hhvm
+
+matrix:
+  allow_failures:
+    - php: hhvm
+
+install:
+  - composer install
+
+script:
+  - phpunit
diff --git a/LICENSE b/LICENSE
@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2016 Voryx LLC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
diff --git a/README.md b/README.md
@@ -0,0 +1,38 @@
+# Stream Component
+
+Provides RxPHP Observables for PHP streams
+
+This library is a wrapper around the [ReactPHP](https://github.com/reactphp/stream) stream library.  It uses the [Voryx event-loop](https://github.com/voryx/event-loop) which behaves like the Javascript event-loop.  ie. You don't need to start it.
+
+
+## Usage
+
+### From File
+```php
+    
+    $source = new \Rx\React\FromFileObservable("example.csv");
+    
+    $source
+        ->cut()
+        ->map(function ($row) {
+            //Convert csv row to an array
+            return str_getcsv($row);
+        })
+        ->map(function (array $row) {
+            //Strip numbers from the first field
+            $row[0] = preg_replace('/\d+/u', '', $row[0]);
+            return $row;
+        })
+        ->subscribe(new \Rx\Observer\CallbackObserver(
+            function ($data) {
+                echo $data[0] . "\n";
+            },
+            function ($e) {
+                echo "error\n";
+            },
+            function () {
+                echo "done\n";
+            }
+        ));
+    
+```
diff --git a/composer.json b/composer.json
@@ -0,0 +1,42 @@
+{
+  "name": "rx/stream",
+  "type": "library",
+  "description": "Async Stream for RxPHP",
+  "keywords": [
+    "rxphp",
+    "reactivex",
+    "react",
+    "reactphp",
+    "stream",
+    "rx.php"
+  ],
+  "license": "MIT",
+  "authors": [
+    {
+      "name": "David Dan",
+      "email": "davidwdan@gmail.com",
+      "role": "Developer"
+    },
+    {
+      "name": "Matt Bonneau",
+      "email": "matt@bonneau.net",
+      "role": "Developer"
+    }
+  ],
+  "autoload": {
+    "psr-4": {
+      "Rx\\React\\": "src/"
+    }
+  },
+  "require": {
+    "voryx/event-loop": "^0.2.0",
+    "react/stream": "^0.4.3",
+    "asm89/rx.php": "dev-master"
+  },
+  "repositories": [
+    {
+      "type": "git",
+      "url": "https://github.com/voryx/Rx.PHP.git"
+    }
+  ]
+}
diff --git a/examples/fromFile/fromFile.php b/examples/fromFile/fromFile.php
@@ -0,0 +1,18 @@
+<?php
+
+include __DIR__ . "/../../vendor/autoload.php";
+
+$source = new \Rx\React\FromFileObservable(__DIR__ . "/../test.csv");
+
+$source
+    ->subscribe(new \Rx\Observer\CallbackObserver(
+        function ($data) {
+            echo $data . "\n";
+        },
+        function ($e) {
+            echo "error\n";
+        },
+        function () {
+            echo "done\n";
+        }
+    ));
diff --git a/examples/fromFile/fromFileWithCut.php b/examples/fromFile/fromFileWithCut.php
@@ -0,0 +1,19 @@
+<?php
+
+include __DIR__ . "/../../vendor/autoload.php";
+
+$source = new \Rx\React\FromFileObservable(__DIR__ . "/../test.csv");
+
+$source
+    ->cut()
+    ->subscribe(new \Rx\Observer\CallbackObserver(
+        function ($data) {
+            echo $data . "\n";
+        },
+        function ($e) {
+            echo "error\n";
+        },
+        function () {
+            echo "done\n";
+        }
+    ));
diff --git a/examples/fromFile/fromFiletoArray.php b/examples/fromFile/fromFiletoArray.php
@@ -0,0 +1,23 @@
+<?php
+
+include __DIR__ . "/../../vendor/autoload.php";
+
+$source = new \Rx\React\FromFileObservable(__DIR__ . "/../test.csv");
+
+$source
+    ->cut()
+    ->map(function ($row) {
+        return str_getcsv($row);
+    })
+    ->toArray()
+    ->subscribe(new \Rx\Observer\CallbackObserver(
+        function (array $array) {
+            var_dump($array);
+        },
+        function ($e) {
+            echo "error\n";
+        },
+        function () {
+            echo "done\n";
+        }
+    ));
diff --git a/examples/test.csv b/examples/test.csv
@@ -0,0 +1,18 @@
+playerID,yearID,stint,teamID,lgID,G,AB,R,H,2B,3B,HR,RBI,SB,CS,BB,SO,IBB,HBP,SH,SF,GIDP
+abercda01,1871,1,TRO,NA,1,4,0,0,0,0,0,0,0,0,0,0,,,,,
+addybo01,1871,1,RC1,NA,25,118,30,32,6,0,0,13,8,1,4,0,,,,,
+allisar01,1871,1,CL1,NA,29,137,28,40,4,5,0,19,3,1,2,5,,,,,
+allisdo01,1871,1,WS3,NA,27,133,28,44,10,2,2,27,1,1,0,2,,,,,
+ansonca01,1871,1,RC1,NA,25,120,29,39,11,3,0,16,6,2,2,1,,,,,
+armstbo01,1871,1,FW1,NA,12,49,9,11,2,1,0,5,0,1,0,1,,,,,
+barkeal01,1871,1,RC1,NA,1,4,0,1,0,0,0,2,0,0,1,0,,,,,
+barnero01,1871,1,BS1,NA,31,157,66,63,10,9,0,34,11,6,13,1,,,,,
+barrebi01,1871,1,FW1,NA,1,5,1,1,1,0,0,1,0,0,0,0,,,,,
+barrofr01,1871,1,BS1,NA,18,86,13,13,2,1,0,11,1,0,0,0,,,,,
+bassjo01,1871,1,CL1,NA,22,89,18,27,1,10,3,18,0,1,3,4,,,,,
+battijo01,1871,1,CL1,NA,1,3,0,0,0,0,0,0,0,0,1,0,,,,,
+bealsto01,1871,1,WS3,NA,10,36,6,7,0,0,0,1,2,0,2,0,,,,,
+beaveed01,1871,1,TRO,NA,3,15,7,6,0,0,0,5,2,0,0,0,,,,,
+bechtge01,1871,1,PH1,NA,20,94,24,33,9,1,1,21,4,0,2,2,,,,,
+bellast01,1871,1,TRO,NA,29,128,26,32,3,3,0,23,4,4,9,2,,,,,
+berkena01,1871,1,PH1,NA,1,4,0,0,0,0,0,0,0,0,0,3,,,,,
diff --git a/phpunit.xml b/phpunit.xml
@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<phpunit
+        backupGlobals="false"
+        backupStaticAttributes="false"
+        colors="true"
+        convertErrorsToExceptions="true"
+        convertNoticesToExceptions="true"
+        convertWarningsToExceptions="true"
+        processIsolation="false"
+        stopOnFailure="false"
+        syntaxCheck="false"
+        bootstrap="tests/bootstrap.php">
+    <testsuites>
+        <testsuite name="Rx/Stream Tests">
+            <directory>tests/</directory>
+        </testsuite>
+    </testsuites>
+
+</phpunit>
diff --git a/src/FromFileObservable.php b/src/FromFileObservable.php
@@ -0,0 +1,73 @@
+<?php
+
+namespace Rx\React;
+
+use React\Stream\Stream;
+use Rx\Disposable\CallbackDisposable;
+use Rx\Observable;
+use Rx\ObserverInterface;
+use Rx\React\Operator\CutOperator;
+use Rx\SchedulerInterface;
+
+class FromFileObservable extends Observable
+{
+    /** @var  string */
+    private $fileName;
+
+    /** @var  string */
+    private $mode;
+
+
+    public function __construct($fileName, $mode = "r")
+    {
+        $this->fileName = $fileName;
+        $this->mode     = $mode;
+    }
+
+    /**
+     * @param ObserverInterface $observer
+     * @param SchedulerInterface|null $scheduler
+     * @return \Rx\Disposable\CompositeDisposable|\Rx\DisposableInterface
+     */
+    public function subscribe(ObserverInterface $observer, SchedulerInterface $scheduler = null)
+    {
+
+        try {
+            $stream = new Stream(fopen($this->fileName, $this->mode), \EventLoop\getLoop());
+
+            $stream->on('data', function ($data) use ($observer) {
+                $observer->onNext($data);
+            });
+
+            $stream->on('error', function ($error) use ($observer) {
+                $e = new \Exception($error);
+                $observer->onError($e);
+            });
+
+            $stream->on('close', function () use ($observer) {
+                $observer->onCompleted();
+            });
+
+            return new CallbackDisposable(function () use ($stream) {
+                $stream->close();
+            });
+
+        } catch (\Exception $e) {
+            $observer->onError($e);
+        }
+
+    }
+
+    /**
+     * Cuts the stream based upon a delimiter.
+     *
+     * @param string $lineEnd
+     * @return Observable\AnonymousObservable
+     */
+    public function cut($lineEnd = PHP_EOL)
+    {
+        return $this->lift(function () use ($lineEnd) {
+            return new CutOperator($lineEnd);
+        });
+    }
+}
diff --git a/src/Operator/CutOperator.php b/src/Operator/CutOperator.php
@@ -0,0 +1,78 @@
+<?php
+
+namespace Rx\React\Operator;
+
+use Rx\Disposable\CompositeDisposable;
+use Rx\ObservableInterface;
+use Rx\Observer\CallbackObserver;
+use Rx\ObserverInterface;
+use Rx\Operator\OperatorInterface;
+use Rx\SchedulerInterface;
+
+/**
+ * Cuts the stream based upon a delimiter.
+ *
+ * Class CutOperator
+ * @package Rx\React\Operator
+ */
+class CutOperator implements OperatorInterface
+{
+
+    private $delimiter;
+
+    public function __construct($delimiter = PHP_EOL)
+    {
+        $this->delimiter = $delimiter;
+    }
+
+    /**
+     * @param \Rx\ObservableInterface $observable
+     * @param \Rx\ObserverInterface $observer
+     * @param \Rx\SchedulerInterface $scheduler
+     * @return \Rx\DisposableInterface
+     */
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
+    {
+        $buffer     = '';
+        $items      = [];
+        $disposable = new CompositeDisposable();
+
+        $onNext = function ($x) use (&$buffer, $observer, $scheduler, &$items, $disposable) {
+            $buffer .= $x;
+            $items  = array_merge($items, explode($this->delimiter, $buffer));
+            $buffer = array_pop($items);
+
+            $action = function ($reschedule) use (&$observer, &$items, &$buffer) {
+
+                if (count($items) === 0) {
+                    return;
+                }
+
+                $value = array_shift($items);
+
+                $observer->onNext($value);
+
+                $reschedule();
+
+            };
+
+            $schedulerDisposable = $scheduler->scheduleRecursive($action);
+
+            $disposable->add($schedulerDisposable);
+        };
+
+        $onCompleted = function () use (&$buffer, $observer) {
+            if (!empty($buffer)) {
+                $observer->onNext($buffer);
+            }
+            $observer->onCompleted();
+        };
+
+        $callbackObserver = new CallbackObserver($onNext, [$observer, "onError"], $onCompleted);
+        $sourceDisposable = $observable->subscribe($callbackObserver);
+
+        $disposable->add($sourceDisposable);
+
+        return $disposable;
+    }
+}
diff --git a/tests/Functional/Observable/FromFileObservableTest.php b/tests/Functional/Observable/FromFileObservableTest.php
diff --git a/tests/Functional/Operator/CutTest.php b/tests/Functional/Operator/CutTest.php
diff --git a/tests/Functional/test.txt b/tests/Functional/test.txt
diff --git a/tests/bootstrap.php b/tests/bootstrap.php
