Merge pull request #17 from dronemill/huge_payloads

Added support for huge payloads

Author: fiddur

composer.json

@@ -11,6 +11,9 @@
     "autoload": {
         "psr-4": {"WebSocket\\": "lib"}
     },
+    "autoload-dev": {
+        "psr-4": {"WebSocket\\Tests\\": "tests"}
+    },
     "require-dev": {
         "phpunit/phpunit": "4.1.*",
         "phpunit/phpunit-selenium": "1.3.3",

examples/echoserver.php

@@ -19,8 +19,11 @@
   $test_id = $server->getPath();
   $test_id = substr($test_id, 1);
 
-  xdebug_start_code_coverage(XDEBUG_CC_UNUSED | XDEBUG_CC_DEAD_CODE);
-  PHPUnit_Extensions_SeleniumCommon_ExitHandler::init();
+  if (function_exists('xdebug_get_code_coverage'))
+    xdebug_start_code_coverage(XDEBUG_CC_UNUSED | XDEBUG_CC_DEAD_CODE);
+
+  if (class_exists('PHPUnit_Extensions_SeleniumCommon_ExitHandler'))
+    PHPUnit_Extensions_SeleniumCommon_ExitHandler::init();
 
   try {
     while(1) {
@@ -56,6 +59,9 @@
 
 
 function save_coverage_data($test_id) {
+
+  if (!function_exists('xdebug_get_code_coverage')) return;
+
   $data = xdebug_get_code_coverage();
   xdebug_stop_code_coverage();
 

lib/Base.php

@@ -4,14 +4,15 @@
 
 class Base {
   protected $socket, $is_connected = false, $is_closing = false, $last_opcode = null,
-    $close_status = null;
+    $close_status = null, $huge_payload = null;
 
   protected static $opcodes = array(
-    'text'   => 1,
-    'binary' => 2,
-    'close'  => 8,
-    'ping'   => 9,
-    'pong'   => 10,
+    'continuation' => 0,
+    'text'         => 1,
+    'binary'       => 2,
+    'close'        => 8,
+    'ping'         => 9,
+    'pong'         => 10,
   );
 
   public function getLastOpcode()  { return $this->last_opcode;  }
@@ -26,20 +27,52 @@ public function setTimeout($timeout) {
     }
   }
 
+  public function setFragmentSize($fragment_size) {
+    $this->options['fragment_size'] = $fragment_size;
+    return $this;
+  }
+
+  public function getFragmentSize() {
+    return $this->options['fragment_size'];
+  }
+
   public function send($payload, $opcode = 'text', $masked = true) {
     if (!$this->is_connected) $this->connect(); /// @todo This is a client function, fixme!
 
     if (!in_array($opcode, array_keys(self::$opcodes))) {
       throw new BadOpcodeException("Bad opcode '$opcode'.  Try 'text' or 'binary'.");
     }
 
+    // record the length of the payload
+    $payload_length = strlen($payload);
+
+    $fragment_cursor = 0;
+    // while we have data to send
+    while ($payload_length > $fragment_cursor) {
+      // get a fragment of the payload
+      $sub_payload = substr($payload, $fragment_cursor, $this->options['fragment_size']);
+
+      // advance the cursor
+      $fragment_cursor += $this->options['fragment_size'];
+
+      // is this the final fragment to send?
+      $final = $payload_length <= $fragment_cursor;
+
+      // send the fragment
+      $this->send_fragment($final, $sub_payload, $opcode, $masked);
+
+      // all fragments after the first will be marked a continuation
+      $opcode = 'continuation';
+    }
+
+  }
+
+  protected function send_fragment($final, $payload, $opcode, $masked) {
     // Binary string for header.
     $frame_head_binstr = '';
 
-
     // Write FIN, final fragment bit.
-    $final = true; /// @todo Support HUGE payloads.
-    $frame_head_binstr .= $final ? '1' : '0';
+    $frame_head_binstr .= (bool) $final ? '1' : '0';
 
     // RSV 1, 2, & 3 false and unused.
     $frame_head_binstr .= '000';
@@ -88,6 +121,16 @@ public function send($payload, $opcode = 'text', $masked = true) {
   public function receive() {
     if (!$this->is_connected) $this->connect(); /// @todo This is a client function, fixme!
 
+    $this->huge_payload = '';
+
+    $response = null;
+    while (is_null($response)) $response = $this->receive_fragment();
+
+    return $response;
+  }
+
+  protected function receive_fragment() {
+
     // Just read the main fragment information first.
     $data = $this->read(2);
 
@@ -107,13 +150,17 @@ public function receive() {
       throw new ConnectionException("Bad opcode in websocket frame: $opcode_int");
     }
     $opcode = $opcode_ints[$opcode_int];
-    $this->last_opcode = $opcode;
+
+    // record the opcode if we are not receiving a continutation fragment
+    if ($opcode !== 'continuation') {
+      $this->last_opcode = $opcode;
+    }
 
     // Masking?
     $mask = (boolean) (ord($data[1]) >> 7);  // Bit 0 in byte 1
 
-    $payload = "";
-    
+    $payload = '';
+
     // Payload length
     $payload_length = (integer) ord($data[1]) & 127; // Bits 1-7 in byte 1
     if ($payload_length > 125) {
@@ -131,7 +178,6 @@ public function receive() {
 
       if ($mask) {
         // Unmask payload.
-        $payload = '';
         for ($i = 0; $i < $payload_length; $i++) $payload .= ($data[$i] ^ $masking_key[$i % 4]);
       }
       else $payload = $data;
@@ -144,16 +190,29 @@ public function receive() {
         $status = bindec(sprintf("%08b%08b", ord($payload[0]), ord($payload[1])));
         $this->close_status = $status;
         $payload = substr($payload, 2);
+
+        if (!$this->is_closing) $this->send($status_bin . 'Close acknowledged: ' . $status, 'close', true); // Respond.
       }
 
       if ($this->is_closing) $this->is_closing = false; // A close response, all done.
-      else $this->send($status_bin . 'Close acknowledged: ' . $status, 'close', true); // Respond.
 
       // And close the socket.
       fclose($this->socket);
       $this->is_connected = false;
     }
 
+    // if this is not the last fragment, then we need to save the payload
+    if (!$final) {
+      $this->huge_payload .= $payload;
+      return null;
+    }
+    // this is the last fragment, and we are processing a huge_payload
+    else if ($this->huge_payload) {
+      // sp we need to retreive the whole payload
+      $payload = $this->huge_payload .= $payload;
+      $this->huge_payload = null;
+    }
+
     return $payload;
   }
 

lib/Client.php

@@ -17,6 +17,9 @@ public function __construct($uri, $options = array()) {
 
     if (!array_key_exists('timeout', $this->options)) $this->options['timeout'] = 5;
 
+    // the fragment size
+    if (!array_key_exists('fragment_size', $this->options)) $this->options['fragment_size'] = 4096;
+
     $this->socket_uri = $uri;
   }
 

lib/Server.php

@@ -15,6 +15,11 @@ class Server extends Base {
    *   - port:     Chose port for listening.
    */
   public function __construct(array $options = array()) {
+
+    // the fragment size
+    if (!array_key_exists('fragment_size', $options)) $options['fragment_size'] = 4096;
+
+
     $this->port = isset($options['port']) ? $options['port'] : 8000;
     $this->options = $options;
 

tests/ClientTracker.php

@@ -0,0 +1,53 @@
+<?php
+
+namespace WebSocket\Tests;
+
+use WebSocket\Client;
+
+/**
+ * A class to track the send and receive operations of the client
+ */
+class ClientTracker extends Client {
+	public $fragment_count;
+
+	public function __construct($uri, $options = array()) {
+		$this->fragment_count = array(
+			'send' => 0,
+			'receive' => 0,
+		);
+
+		return parent::__construct($uri, $options);
+	}
+
+	/**
+	 * Reset the send counter
+	 */
+	public function send($payload, $opcode = 'text', $masked = true) {
+		$this->fragment_count['send'] = 0;
+		return parent::send($payload, $opcode, $masked);
+	}
+
+	/**
+	 * Increment the fragment send counter
+	 */
+	protected function send_fragment($final, $payload, $opcode, $masked) {
+		$this->fragment_count['send']++;
+		return parent::send_fragment($final, $payload, $opcode, $masked);
+	}
+
+	/**
+	 * Reset the receive counter
+	 */
+	public function receive() {
+		$this->fragment_count['receive'] = 0;
+		return parent::receive();
+	}
+
+	/**
+	 * Increment the fragment recieve counter
+	 */
+	protected function receive_fragment() {
+		$this->fragment_count['receive']++;
+		return parent::receive_fragment();
+	}
+}

tests/unit/ClientTest.php

@@ -1,6 +1,7 @@
 <?php
 
 use WebSocket\Client;
+use WebSocket\Tests\ClientTracker;
 
 class WebSocketTest extends PHPUnit_Framework_TestCase {
   protected static $port;
@@ -69,11 +70,28 @@ public function testInstantiation() {
     $this->assertInstanceOf('WebSocket\Client', $ws);
   }
 
+  /**
+   * Data provider for testEcho
+   *
+   * @return  array of data
+   */
+  public function dataLengthProvider() {
+    return array(
+      array(8),
+      array(126),
+      array(127),
+      array(128),
+      array(65000),
+      array(66000),
+    );
+  }
+
   /**
    * @dataProvider dataLengthProvider
    */
   public function testEcho($data_length) {
-    $ws = new Client('ws://localhost:' . self::$port . '/' . $this->test_id);
+    $ws = new ClientTracker('ws://localhost:' . self::$port . '/' . $this->test_id);
+    $ws->setFragmentSize(rand(10,512));
 
     $greeting = '';
     for ($i = 0; $i < $data_length; $i++) $greeting .= 'o';
@@ -82,6 +100,8 @@ public function testEcho($data_length) {
     $response = $ws->receive();
 
     $this->assertEquals($greeting, $response);
+    $this->assertEquals($ws->fragment_count['send'], ceil($data_length / $ws->getFragmentSize()));
+    $this->assertEquals($ws->fragment_count['receive'], ceil($data_length / 4096)); // the server sends with size 4096
   }
 
   public function testBasicAuth() {
@@ -98,17 +118,6 @@ public function testBasicAuth() {
     $this->assertEquals("Basic Sm9obkRvZTplb0RuaG9K - $greeting", $response);
   }
 
-  public function dataLengthProvider() {
-    return array(
-      array(8),
-      array(126),
-      array(127),
-      array(128),
-      array(65000),
-      array(66000),
-    );
-  }
-
   public function testOrgEchoTwice() {
     $ws = new Client('ws://localhost:' . self::$port . '/' . $this->test_id);
 
@@ -290,4 +299,10 @@ public function testSendBadOpcode() {
     $ws = new Client('ws://localhost:' . self::$port);
     $ws->send('foo', 'bad_opcode');
   }
+
+  public function testSetFragmentSize() {
+    $ws = new Client('ws://localhost:' . self::$port);
+    $size = $ws->setFragmentSize(123)->getFragmentSize();
+    $this->assertSame(123, $size);
+  }
 }