Fix http json event double encoding

Rxnet · Aug 16, 2017 · ac1aa73 · ac1aa73
1 parent 761da2c
commit ac1aa73
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 15 deletions.
diff --git a/src/Rxnet/EventStore/EventStore.php b/src/Rxnet/EventStore/EventStore.php
@@ -100,7 +100,7 @@ public function __construct(LoopInterface $loop = null, Dns $dns = null, Tcp $tc
      * @param int $heartBeatRate in milliseconds
      * @return Observable\AnonymousObservable
      */
-    public function connect($dsn = 'tcp://admin:changeit@localhost:1113', $connectTimeout = 1000, $heartBeatRate = 5000)
+    public function connect($dsn = 'tcp://admin:changeit@127.0.0.1:1113', $connectTimeout = 1000, $heartBeatRate = 5000)
     {
         // connector compatibility
         $connectTimeout = ($connectTimeout > 0) ? $connectTimeout / 1000 : 0;
@@ -636,4 +636,4 @@ protected function readEvents(Message $query, $messageType)
             });
 
     }
-}
+}
diff --git a/src/Rxnet/EventStore/HttpEventStore.php b/src/Rxnet/EventStore/HttpEventStore.php
@@ -53,7 +53,6 @@ public function write($streamId, $events, $expectedVersion = null)
             $headers['ES-ExpectedVersion'] = $expectedVersion;
         }
 
-
         $body = json_encode($json);
         $headers['Length'] = strlen($body);
         $options = compact('headers', 'body');

diff --git a/src/Rxnet/EventStore/NewEvent/JsonEvent.php b/src/Rxnet/EventStore/NewEvent/JsonEvent.php
@@ -58,4 +58,13 @@ public function getMessage()
     {
         return $this->message;
     }
+    public function toArray()
+    {
+        return [
+            'eventId' => $this->getId(),
+            'eventType' => $this->getType(),
+            'data' => json_decode($this->getData(), true),
+            'metadata' => json_decode($this->getMetaData(), true)
+        ];
+    }
 }
diff --git a/usage/http-write.php b/usage/http-write.php
@@ -9,13 +9,16 @@
 
 $eventStore = new \Rxnet\EventStore\HttpEventStore();
 
-\Rx\Observable::interval(10)
+\Rx\Observable::interval(1000)
     ->flatMap(
         function ($i) use ($eventStore) {
-            $uid = \Ramsey\Uuid\Uuid::uuid3(\Ramsey\Uuid\Uuid::NAMESPACE_DNS, 'test.fr');
-            $event = new JsonEvent('/truc/chose', ['i' => $i], 'test.fr');
+            $event = new JsonEvent('/truc/chose', [
+                "crypto" => "btc",
+                "user_id" => "21433R4G523TO",
+                "wallet_id" => "PIH21B4T93VB5T9G7V"
+            ]);
 
-            return $eventStore->write('domain-test3.fr', $event);
+            return $eventStore->write('crypto-exchange-rate', $event);
         }
     )
     ->subscribe(
@@ -24,11 +27,11 @@ function ($eventsCompleted) {
 
                 gc_collect_cycles();
                 $memory = memory_get_usage(true) / 1024 / 1024;
-                echo $eventsCompleted." {$memory}Mb \n";
+                echo $eventsCompleted . " {$memory}Mb \n";
                 //echo "Last event number {$eventsCompleted->getLastEventNumber()} on commit position {$eventsCompleted->getCommitPosition()} {$memory}Mb \n";
             }
         ),
         new EventLoopScheduler(EventLoop::getLoop())
     );
 
-EventLoop::getLoop()->run();
+EventLoop::getLoop()->run();
diff --git a/usage/persistent.php b/usage/persistent.php
@@ -6,11 +6,12 @@
 $eventStore = new \Rxnet\EventStore\EventStore();
 \Rxnet\await($eventStore->connect());
 
-$eventStore->persistentSubscription('dropcatch', 'test')
+$eventStore->persistentSubscription('crypto', 'test')
     ->flatMap(function (AcknowledgeableEventRecord $record) {
         echo "received {$record->getId()}  {$record->getNumber()}@{$record->getStreamId()} {$record->getType()}\n";
-        //return $record->ack();
-        return $record->nack($record::NACK_ACTION_PARK, 'oops');
+        var_dump($record->getData());
+        return $record->ack();
+        //return $record->nack($record::NACK_ACTION_PARK, 'oops');
     })
     ->subscribeCallback(null, function (\Exception $e) {
         echo $e->getMessage();

diff --git a/usage/write.php b/usage/write.php
@@ -19,8 +19,10 @@
     ->flatMap(
         function ($i) use ($eventStore) {
             echo '.';
-            $event = new JsonEvent('/truc/chose', ['i' => $i]);
-            return $eventStore->write('domain-test-1.fr', [$event]);
+            $event = new JsonEvent('/truc/chose', [ "crypto"=> "btc",
+      "user_id"=> "21433R4G523TO",
+      "wallet_id"=> "PIH21B4T93VB5T9G7V"]);
+            return $eventStore->write('test-test-1.fr', [$event]);
         }
     )
     ->subscribe(
@@ -34,4 +36,4 @@ function (WriteEventsCompleted $eventsCompleted) {
         new EventLoopScheduler(EventLoop::getLoop())
     );
 
-EventLoop::getLoop()->run();
+EventLoop::getLoop()->run();