Prototype

src/Riak/Client/Command/DataType/FetchMap.php

@@ -40,14 +40,19 @@ public function __construct(RiakLocation $location = null, array $options = [])
      */
     public function execute(RiakCluster $cluster)
     {
-        $config    = $cluster->getRiakConfig();
-        $converter = $config->getCrdtResponseConverter();
-        $operation = new FetchMapOperation($converter, $this->location, $this->options);
+        $operation = $this->createOperation($cluster);
         $response  = $cluster->execute($operation);
 
         return $response;
     }
 
+    public function createOperation(RiakCluster $cluster)
+    {
+        $config    = $cluster->getRiakConfig();
+        $converter = $config->getCrdtResponseConverter();
+        return new FetchMapOperation($converter, $this->location, $this->options);
+    }
+
     /**
      * @param \Riak\Client\Core\Query\RiakLocation $location
      * @param array                                $options

src/Riak/Client/Core/Operation/DataType/FetchDataTypeOperation.php

@@ -48,12 +48,9 @@ public function __construct(CrdtResponseConverter $converter, RiakLocation $loca
      */
     public function execute(RiakTransport $adapter)
     {
-        $getRequest  = $this->createGetRequest();
+        $getRequest  = $this->createRequest();
         $getResponse = $adapter->send($getRequest);
-        $datatype    = $this->converter->convert($getResponse);
-        $response    = $this->createDataTypeResponse($datatype, $getResponse->context);
-
-        return $response;
+        return $this->createResponse($getResponse);
     }
 
     /**
@@ -75,6 +72,19 @@ private function createGetRequest()
         return $request;
     }
 
+    public function createRequest()
+    {
+        return $this->createGetRequest();
+    }
+
+    public function createResponse($response)
+    {
+        $datatype    = $this->converter->convert($response);
+        $response    = $this->createDataTypeResponse($datatype, $response->context);
+
+        return $response;
+    }
+
     /**
      * @param \Riak\Client\Core\Query\Crdt\DataType $datatype
      * @param string                                $context

src/Riak/Client/Core/RiakCluster.php

@@ -97,15 +97,25 @@ public function getRiakConfig()
      * @return \Riak\Client\RiakResponse
      */
     public function execute(RiakOperation $operation)
+    {
+        $node = $this->selectNode();
+        return $node->execute($operation);
+    }
+
+    public function batch(array $operations)
+    {
+        $node = $this->selectNode();
+        return $node->batch($operations);
+    }
+
+    private function selectNode()
     {
         if (empty($this->nodes)) {
             throw new RiakException('There are no nodes in the cluster.');
         }
 
         $size  = count($this->nodes);
         $index = mt_rand(0, $size - 1);
-        $node  = $this->nodes[$index];
-
-        return $node->execute($operation);
+        return $this->nodes[$index];
     }
 }

src/Riak/Client/Core/RiakHttpTransport.php

@@ -91,4 +91,30 @@ public function send(Request $request)
             throw RiakTransportException::httpRequestException($exc);
         }
     }
+
+    public function batch(array $requests)
+    {
+        $strategies = [];
+        foreach ($requests as $key => $req) {
+            $strategies[$key] = $this->createAdapterStrategyFor($req);
+        }
+
+        $strategiesRequests = [];
+        foreach ($strategies as $key => $strategy) {
+            $strategiesRequests[$key] = $strategy->buildRequest($requests[$key]);
+        }
+
+        $results = \GuzzleHttp\Pool::batch($this->client, $strategiesRequests);
+
+        $strategiesResponses = [];
+        foreach($strategiesRequests as $key => $strategyRequest) {
+            $strategiesResponses[$key] = $results->getResult($strategyRequest);
+        }
+
+        $responses = [];
+        foreach ($strategiesResponses as $key => $strategyResponse) {
+            $responses[$key] = $strategies[$key]->buildResponse($strategiesResponses[$key]);
+        }
+        return $responses;
+    }
 }

src/Riak/Client/Core/RiakNode.php

@@ -40,4 +40,22 @@ public function execute(RiakOperation $operation)
     {
         return $operation->execute($this->adapter);
     }
+
+    public function batch(array $operations)
+    {
+        $requests = array_map(
+            function ($operation) {
+                return $operation->createRequest();
+            },
+            $operations
+        );
+
+        $adapterResponses = $this->adapter->batch($requests);
+
+        $responses = [];
+        foreach ($adapterResponses as $key => $resp) {
+            $responses[$key] = $operations[$key]->createResponse($resp);
+        }
+        return $responses;
+    }
 }

src/Riak/Client/Core/Transport/Http/DataType/HttpGet.php

@@ -1,5 +1,5 @@
 <?php
-
+ 
 namespace Riak\Client\Core\Transport\Http\DataType;
 
 use Riak\Client\Core\Message\Request;
@@ -64,8 +64,7 @@ private function createHttpRequest(GetRequest $getRequest)
      */
     public function send(Request $request)
     {
-        $response    = new GetResponse();
-        $httpRequest = $this->createHttpRequest($request);
+        $httpRequest = $this->buildRequest($request);
 
         try {
             $httpResponse = $this->client->send($httpRequest);
@@ -78,6 +77,19 @@ public function send(Request $request)
             throw $e;
         }
 
+        return $this->buildResponse($httpResponse);
+    }
+
+    public function buildRequest(Request $request)
+    {
+        return $this->createHttpRequest($request);
+    }
+
+    public function buildResponse($httpResponse)
+    {
+        $response = new GetResponse();
+
+        $code = $httpResponse->getStatusCode();        
         if ( ! isset($this->validResponseCodes[$code])) {
             throw RiakTransportException::unexpectedStatusCode($code);
         }
@@ -91,6 +103,6 @@ public function send(Request $request)
         $response->context = $context;
         $response->value   = $this->opConverter->fromArray($type, $value);
 
-        return $response;
+        return $response;        
     }
 }

src/Riak/Client/RiakClient.php

@@ -60,4 +60,16 @@ public function execute(RiakCommand $command)
     {
         return $command->execute($this->cluster);
     }
+
+    public function batch(array $commands)
+    {
+        $cluster = $this->cluster;
+        $operations = array_map(
+            function($command) use ($cluster) {
+                return $command->createOperation($cluster);
+            },
+            $commands
+        );
+        return $this->cluster->batch($operations);
+    }
 }