Thrift now a TLP - INFRA-3116
git-svn-id: https://svn.apache.org/repos/asf/thrift/branches/0.1.x@1028168 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/lib/php/src/transport/TBufferedTransport.php b/lib/php/src/transport/TBufferedTransport.php
new file mode 100644
index 0000000..cfae767
--- /dev/null
+++ b/lib/php/src/transport/TBufferedTransport.php
@@ -0,0 +1,163 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Buffered transport. Stores data to an internal buffer that it doesn't
+ * actually write out until flush is called. For reading, we do a greedy
+ * read and then serve data out of the internal buffer.
+ *
+ * @package thrift.transport
+ */
+class TBufferedTransport extends TTransport {
+
+ /**
+ * Constructor. Creates a buffered transport around an underlying transport
+ */
+ public function __construct($transport=null, $rBufSize=512, $wBufSize=512) {
+ $this->transport_ = $transport;
+ $this->rBufSize_ = $rBufSize;
+ $this->wBufSize_ = $wBufSize;
+ }
+
+ /**
+ * The underlying transport
+ *
+ * @var TTransport
+ */
+ protected $transport_ = null;
+
+ /**
+ * The receive buffer size
+ *
+ * @var int
+ */
+ protected $rBufSize_ = 512;
+
+ /**
+ * The write buffer size
+ *
+ * @var int
+ */
+ protected $wBufSize_ = 512;
+
+ /**
+ * The write buffer.
+ *
+ * @var string
+ */
+ protected $wBuf_ = '';
+
+ /**
+ * The read buffer.
+ *
+ * @var string
+ */
+ protected $rBuf_ = '';
+
+ public function isOpen() {
+ return $this->transport_->isOpen();
+ }
+
+ public function open() {
+ $this->transport_->open();
+ }
+
+ public function close() {
+ $this->transport_->close();
+ }
+
+ public function putBack($data) {
+ if (strlen($this->rBuf_) === 0) {
+ $this->rBuf_ = $data;
+ } else {
+ $this->rBuf_ = ($data . $this->rBuf_);
+ }
+ }
+
+ /**
+ * The reason that we customize readAll here is that the majority of PHP
+ * streams are already internally buffered by PHP. The socket stream, for
+ * example, buffers internally and blocks if you call read with $len greater
+ * than the amount of data available, unlike recv() in C.
+ *
+ * Therefore, use the readAll method of the wrapped transport inside
+ * the buffered readAll.
+ */
+ public function readAll($len) {
+ $have = strlen($this->rBuf_);
+ if ($have == 0) {
+ $data = $this->transport_->readAll($len);
+ } else if ($have < $len) {
+ $data = $this->rBuf_;
+ $this->rBuf_ = '';
+ $data .= $this->transport_->readAll($len - $have);
+ } else if ($have == $len) {
+ $data = $this->rBuf_;
+ $this->rBuf_ = '';
+ } else if ($have > $len) {
+ $data = substr($this->rBuf_, 0, $len);
+ $this->rBuf_ = substr($this->rBuf_, $len);
+ }
+ return $data;
+ }
+
+ public function read($len) {
+ if (strlen($this->rBuf_) === 0) {
+ $this->rBuf_ = $this->transport_->read($this->rBufSize_);
+ }
+
+ if (strlen($this->rBuf_) <= $len) {
+ $ret = $this->rBuf_;
+ $this->rBuf_ = '';
+ return $ret;
+ }
+
+ $ret = substr($this->rBuf_, 0, $len);
+ $this->rBuf_ = substr($this->rBuf_, $len);
+ return $ret;
+ }
+
+ public function write($buf) {
+ $this->wBuf_ .= $buf;
+ if (strlen($this->wBuf_) >= $this->wBufSize_) {
+ $out = $this->wBuf_;
+
+ // Note that we clear the internal wBuf_ prior to the underlying write
+ // to ensure we're in a sane state (i.e. internal buffer cleaned)
+ // if the underlying write throws up an exception
+ $this->wBuf_ = '';
+ $this->transport_->write($out);
+ }
+ }
+
+ public function flush() {
+ if (strlen($this->wBuf_) > 0) {
+ $this->transport_->write($this->wBuf_);
+ $this->wBuf_ = '';
+ }
+ $this->transport_->flush();
+ }
+
+}
+
+?>
diff --git a/lib/php/src/transport/TFramedTransport.php b/lib/php/src/transport/TFramedTransport.php
new file mode 100644
index 0000000..dc57392
--- /dev/null
+++ b/lib/php/src/transport/TFramedTransport.php
@@ -0,0 +1,179 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Framed transport. Writes and reads data in chunks that are stamped with
+ * their length.
+ *
+ * @package thrift.transport
+ */
+class TFramedTransport extends TTransport {
+
+ /**
+ * Underlying transport object.
+ *
+ * @var TTransport
+ */
+ private $transport_;
+
+ /**
+ * Buffer for read data.
+ *
+ * @var string
+ */
+ private $rBuf_;
+
+ /**
+ * Buffer for queued output data
+ *
+ * @var string
+ */
+ private $wBuf_;
+
+ /**
+ * Whether to frame reads
+ *
+ * @var bool
+ */
+ private $read_;
+
+ /**
+ * Whether to frame writes
+ *
+ * @var bool
+ */
+ private $write_;
+
+ /**
+ * Constructor.
+ *
+ * @param TTransport $transport Underlying transport
+ */
+ public function __construct($transport=null, $read=true, $write=true) {
+ $this->transport_ = $transport;
+ $this->read_ = $read;
+ $this->write_ = $write;
+ }
+
+ public function isOpen() {
+ return $this->transport_->isOpen();
+ }
+
+ public function open() {
+ $this->transport_->open();
+ }
+
+ public function close() {
+ $this->transport_->close();
+ }
+
+ /**
+ * Reads from the buffer. When more data is required reads another entire
+ * chunk and serves future reads out of that.
+ *
+ * @param int $len How much data
+ */
+ public function read($len) {
+ if (!$this->read_) {
+ return $this->transport_->read($len);
+ }
+
+ if (strlen($this->rBuf_) === 0) {
+ $this->readFrame();
+ }
+
+ // Just return full buff
+ if ($len >= strlen($this->rBuf_)) {
+ $out = $this->rBuf_;
+ $this->rBuf_ = null;
+ return $out;
+ }
+
+ // Return substr
+ $out = substr($this->rBuf_, 0, $len);
+ $this->rBuf_ = substr($this->rBuf_, $len);
+ return $out;
+ }
+
+ /**
+ * Put previously read data back into the buffer
+ *
+ * @param string $data data to return
+ */
+ public function putBack($data) {
+ if (strlen($this->rBuf_) === 0) {
+ $this->rBuf_ = $data;
+ } else {
+ $this->rBuf_ = ($data . $this->rBuf_);
+ }
+ }
+
+ /**
+ * Reads a chunk of data into the internal read buffer.
+ */
+ private function readFrame() {
+ $buf = $this->transport_->readAll(4);
+ $val = unpack('N', $buf);
+ $sz = $val[1];
+
+ $this->rBuf_ = $this->transport_->readAll($sz);
+ }
+
+ /**
+ * Writes some data to the pending output buffer.
+ *
+ * @param string $buf The data
+ * @param int $len Limit of bytes to write
+ */
+ public function write($buf, $len=null) {
+ if (!$this->write_) {
+ return $this->transport_->write($buf, $len);
+ }
+
+ if ($len !== null && $len < strlen($buf)) {
+ $buf = substr($buf, 0, $len);
+ }
+ $this->wBuf_ .= $buf;
+ }
+
+ /**
+ * Writes the output buffer to the stream in the format of a 4-byte length
+ * followed by the actual data.
+ */
+ public function flush() {
+ if (!$this->write_) {
+ return $this->transport_->flush();
+ }
+
+ $out = pack('N', strlen($this->wBuf_));
+ $out .= $this->wBuf_;
+
+ // Note that we clear the internal wBuf_ prior to the underlying write
+ // to ensure we're in a sane state (i.e. internal buffer cleaned)
+ // if the underlying write throws up an exception
+ $this->wBuf_ = '';
+ $this->transport_->write($out);
+ $this->transport_->flush();
+ }
+
+}
diff --git a/lib/php/src/transport/THttpClient.php b/lib/php/src/transport/THttpClient.php
new file mode 100644
index 0000000..224d403
--- /dev/null
+++ b/lib/php/src/transport/THttpClient.php
@@ -0,0 +1,202 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * HTTP client for Thrift
+ *
+ * @package thrift.transport
+ */
+class THttpClient extends TTransport {
+
+ /**
+ * The host to connect to
+ *
+ * @var string
+ */
+ protected $host_;
+
+ /**
+ * The port to connect on
+ *
+ * @var int
+ */
+ protected $port_;
+
+ /**
+ * The URI to request
+ *
+ * @var string
+ */
+ protected $uri_;
+
+ /**
+ * The scheme to use for the request, i.e. http, https
+ *
+ * @var string
+ */
+ protected $scheme_;
+
+ /**
+ * Buffer for the HTTP request data
+ *
+ * @var string
+ */
+ protected $buf_;
+
+ /**
+ * Input socket stream.
+ *
+ * @var resource
+ */
+ protected $handle_;
+
+ /**
+ * Read timeout
+ *
+ * @var float
+ */
+ protected $timeout_;
+
+ /**
+ * Make a new HTTP client.
+ *
+ * @param string $host
+ * @param int $port
+ * @param string $uri
+ */
+ public function __construct($host, $port=80, $uri='', $scheme = 'http') {
+ if ((strlen($uri) > 0) && ($uri{0} != '/')) {
+ $uri = '/'.$uri;
+ }
+ $this->scheme_ = $scheme;
+ $this->host_ = $host;
+ $this->port_ = $port;
+ $this->uri_ = $uri;
+ $this->buf_ = '';
+ $this->handle_ = null;
+ $this->timeout_ = null;
+ }
+
+ /**
+ * Set read timeout
+ *
+ * @param float $timeout
+ */
+ public function setTimeoutSecs($timeout) {
+ $this->timeout_ = $timeout;
+ }
+
+ /**
+ * Whether this transport is open.
+ *
+ * @return boolean true if open
+ */
+ public function isOpen() {
+ return true;
+ }
+
+ /**
+ * Open the transport for reading/writing
+ *
+ * @throws TTransportException if cannot open
+ */
+ public function open() {}
+
+ /**
+ * Close the transport.
+ */
+ public function close() {
+ if ($this->handle_) {
+ @fclose($this->handle_);
+ $this->handle_ = null;
+ }
+ }
+
+ /**
+ * Read some data into the array.
+ *
+ * @param int $len How much to read
+ * @return string The data that has been read
+ * @throws TTransportException if cannot read any more data
+ */
+ public function read($len) {
+ $data = @fread($this->handle_, $len);
+ if ($data === FALSE || $data === '') {
+ $md = stream_get_meta_data($this->handle_);
+ if ($md['timed_out']) {
+ throw new TTransportException('THttpClient: timed out reading '.$len.' bytes from '.$this->host_.':'.$this->port_.'/'.$this->uri_, TTransportException::TIMED_OUT);
+ } else {
+ throw new TTransportException('THttpClient: Could not read '.$len.' bytes from '.$this->host_.':'.$this->port_.'/'.$this->uri_, TTransportException::UNKNOWN);
+ }
+ }
+ return $data;
+ }
+
+ /**
+ * Writes some data into the pending buffer
+ *
+ * @param string $buf The data to write
+ * @throws TTransportException if writing fails
+ */
+ public function write($buf) {
+ $this->buf_ .= $buf;
+ }
+
+ /**
+ * Opens and sends the actual request over the HTTP connection
+ *
+ * @throws TTransportException if a writing error occurs
+ */
+ public function flush() {
+ // God, PHP really has some esoteric ways of doing simple things.
+ $host = $this->host_.($this->port_ != 80 ? ':'.$this->port_ : '');
+
+ $headers = array('Host: '.$host,
+ 'Accept: application/x-thrift',
+ 'User-Agent: PHP/THttpClient',
+ 'Content-Type: application/x-thrift',
+ 'Content-Length: '.strlen($this->buf_));
+
+ $options = array('method' => 'POST',
+ 'header' => implode("\r\n", $headers),
+ 'max_redirects' => 1,
+ 'content' => $this->buf_);
+ if ($this->timeout_ > 0) {
+ $options['timeout'] = $this->timeout_;
+ }
+ $this->buf_ = '';
+
+ $contextid = stream_context_create(array('http' => $options));
+ $this->handle_ = @fopen($this->scheme_.'://'.$host.$this->uri_, 'r', false, $contextid);
+
+ // Connect failed?
+ if ($this->handle_ === FALSE) {
+ $this->handle_ = null;
+ $error = 'THttpClient: Could not connect to '.$host.$this->uri_;
+ throw new TTransportException($error, TTransportException::NOT_OPEN);
+ }
+ }
+
+}
+
+?>
diff --git a/lib/php/src/transport/TMemoryBuffer.php b/lib/php/src/transport/TMemoryBuffer.php
new file mode 100644
index 0000000..01eb0f5
--- /dev/null
+++ b/lib/php/src/transport/TMemoryBuffer.php
@@ -0,0 +1,84 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * A memory buffer is a tranpsort that simply reads from and writes to an
+ * in-memory string buffer. Anytime you call write on it, the data is simply
+ * placed into a buffer, and anytime you call read, data is read from that
+ * buffer.
+ *
+ * @package thrift.transport
+ */
+class TMemoryBuffer extends TTransport {
+
+ /**
+ * Constructor. Optionally pass an initial value
+ * for the buffer.
+ */
+ public function __construct($buf = '') {
+ $this->buf_ = $buf;
+ }
+
+ protected $buf_ = '';
+
+ public function isOpen() {
+ return true;
+ }
+
+ public function open() {}
+
+ public function close() {}
+
+ public function write($buf) {
+ $this->buf_ .= $buf;
+ }
+
+ public function read($len) {
+ if (strlen($this->buf_) === 0) {
+ throw new TTransportException('TMemoryBuffer: Could not read ' .
+ $len . ' bytes from buffer.',
+ TTransportException::UNKNOWN);
+ }
+
+ if (strlen($this->buf_) <= $len) {
+ $ret = $this->buf_;
+ $this->buf_ = '';
+ return $ret;
+ }
+
+ $ret = substr($this->buf_, 0, $len);
+ $this->buf_ = substr($this->buf_, $len);
+
+ return $ret;
+ }
+
+ function getBuffer() {
+ return $this->buf_;
+ }
+
+ public function available() {
+ return strlen($this->buf_);
+ }
+}
+
+?>
diff --git a/lib/php/src/transport/TNullTransport.php b/lib/php/src/transport/TNullTransport.php
new file mode 100644
index 0000000..bada5df
--- /dev/null
+++ b/lib/php/src/transport/TNullTransport.php
@@ -0,0 +1,48 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Transport that only accepts writes and ignores them.
+ * This is useful for measuring the serialized size of structures.
+ *
+ * @package thrift.transport
+ */
+class TNullTransport extends TTransport {
+
+ public function isOpen() {
+ return true;
+ }
+
+ public function open() {}
+
+ public function close() {}
+
+ public function read($len) {
+ throw new TTransportException("Can't read from TNullTransport.");
+ }
+
+ public function write($buf) {}
+
+}
+
+?>
diff --git a/lib/php/src/transport/TPhpStream.php b/lib/php/src/transport/TPhpStream.php
new file mode 100644
index 0000000..3a1c80b
--- /dev/null
+++ b/lib/php/src/transport/TPhpStream.php
@@ -0,0 +1,111 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Php stream transport. Reads to and writes from the php standard streams
+ * php://input and php://output
+ *
+ * @package thrift.transport
+ */
+class TPhpStream extends TTransport {
+
+ const MODE_R = 1;
+ const MODE_W = 2;
+
+ private $inStream_ = null;
+
+ private $outStream_ = null;
+
+ private $read_ = false;
+
+ private $write_ = false;
+
+ public function __construct($mode) {
+ $this->read_ = $mode & self::MODE_R;
+ $this->write_ = $mode & self::MODE_W;
+ }
+
+ public function open() {
+ if ($this->read_) {
+ $this->inStream_ = @fopen(self::inStreamName(), 'r');
+ if (!is_resource($this->inStream_)) {
+ throw new TException('TPhpStream: Could not open php://input');
+ }
+ }
+ if ($this->write_) {
+ $this->outStream_ = @fopen('php://output', 'w');
+ if (!is_resource($this->outStream_)) {
+ throw new TException('TPhpStream: Could not open php://output');
+ }
+ }
+ }
+
+ public function close() {
+ if ($this->read_) {
+ @fclose($this->inStream_);
+ $this->inStream_ = null;
+ }
+ if ($this->write_) {
+ @fclose($this->outStream_);
+ $this->outStream_ = null;
+ }
+ }
+
+ public function isOpen() {
+ return
+ (!$this->read_ || is_resource($this->inStream_)) &&
+ (!$this->write_ || is_resource($this->outStream_));
+ }
+
+ public function read($len) {
+ $data = @fread($this->inStream_, $len);
+ if ($data === FALSE || $data === '') {
+ throw new TException('TPhpStream: Could not read '.$len.' bytes');
+ }
+ return $data;
+ }
+
+ public function write($buf) {
+ while (strlen($buf) > 0) {
+ $got = @fwrite($this->outStream_, $buf);
+ if ($got === 0 || $got === FALSE) {
+ throw new TException('TPhpStream: Could not write '.strlen($buf).' bytes');
+ }
+ $buf = substr($buf, $got);
+ }
+ }
+
+ public function flush() {
+ @fflush($this->outStream_);
+ }
+
+ private static function inStreamName() {
+ if (php_sapi_name() == 'cli') {
+ return 'php://stdin';
+ }
+ return 'php://input';
+ }
+
+}
+
+?>
diff --git a/lib/php/src/transport/TSocket.php b/lib/php/src/transport/TSocket.php
new file mode 100644
index 0000000..ba3a631
--- /dev/null
+++ b/lib/php/src/transport/TSocket.php
@@ -0,0 +1,312 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Sockets implementation of the TTransport interface.
+ *
+ * @package thrift.transport
+ */
+class TSocket extends TTransport {
+
+ /**
+ * Handle to PHP socket
+ *
+ * @var resource
+ */
+ private $handle_ = null;
+
+ /**
+ * Remote hostname
+ *
+ * @var string
+ */
+ protected $host_ = 'localhost';
+
+ /**
+ * Remote port
+ *
+ * @var int
+ */
+ protected $port_ = '9090';
+
+ /**
+ * Send timeout in milliseconds
+ *
+ * @var int
+ */
+ private $sendTimeout_ = 100;
+
+ /**
+ * Recv timeout in milliseconds
+ *
+ * @var int
+ */
+ private $recvTimeout_ = 750;
+
+ /**
+ * Is send timeout set?
+ *
+ * @var bool
+ */
+ private $sendTimeoutSet_ = FALSE;
+
+ /**
+ * Persistent socket or plain?
+ *
+ * @var bool
+ */
+ private $persist_ = FALSE;
+
+ /**
+ * Debugging on?
+ *
+ * @var bool
+ */
+ protected $debug_ = FALSE;
+
+ /**
+ * Debug handler
+ *
+ * @var mixed
+ */
+ protected $debugHandler_ = null;
+
+ /**
+ * Socket constructor
+ *
+ * @param string $host Remote hostname
+ * @param int $port Remote port
+ * @param bool $persist Whether to use a persistent socket
+ * @param string $debugHandler Function to call for error logging
+ */
+ public function __construct($host='localhost',
+ $port=9090,
+ $persist=FALSE,
+ $debugHandler=null) {
+ $this->host_ = $host;
+ $this->port_ = $port;
+ $this->persist_ = $persist;
+ $this->debugHandler_ = $debugHandler ? $debugHandler : 'error_log';
+ }
+
+ /**
+ * Sets the send timeout.
+ *
+ * @param int $timeout Timeout in milliseconds.
+ */
+ public function setSendTimeout($timeout) {
+ $this->sendTimeout_ = $timeout;
+ }
+
+ /**
+ * Sets the receive timeout.
+ *
+ * @param int $timeout Timeout in milliseconds.
+ */
+ public function setRecvTimeout($timeout) {
+ $this->recvTimeout_ = $timeout;
+ }
+
+ /**
+ * Sets debugging output on or off
+ *
+ * @param bool $debug
+ */
+ public function setDebug($debug) {
+ $this->debug_ = $debug;
+ }
+
+ /**
+ * Get the host that this socket is connected to
+ *
+ * @return string host
+ */
+ public function getHost() {
+ return $this->host_;
+ }
+
+ /**
+ * Get the remote port that this socket is connected to
+ *
+ * @return int port
+ */
+ public function getPort() {
+ return $this->port_;
+ }
+
+ /**
+ * Tests whether this is open
+ *
+ * @return bool true if the socket is open
+ */
+ public function isOpen() {
+ return is_resource($this->handle_);
+ }
+
+ /**
+ * Connects the socket.
+ */
+ public function open() {
+
+ if ($this->persist_) {
+ $this->handle_ = @pfsockopen($this->host_,
+ $this->port_,
+ $errno,
+ $errstr,
+ $this->sendTimeout_/1000.0);
+ } else {
+ $this->handle_ = @fsockopen($this->host_,
+ $this->port_,
+ $errno,
+ $errstr,
+ $this->sendTimeout_/1000.0);
+ }
+
+ // Connect failed?
+ if ($this->handle_ === FALSE) {
+ $error = 'TSocket: Could not connect to '.$this->host_.':'.$this->port_.' ('.$errstr.' ['.$errno.'])';
+ if ($this->debug_) {
+ call_user_func($this->debugHandler_, $error);
+ }
+ throw new TException($error);
+ }
+
+ stream_set_timeout($this->handle_, 0, $this->sendTimeout_*1000);
+ $this->sendTimeoutSet_ = TRUE;
+ }
+
+ /**
+ * Closes the socket.
+ */
+ public function close() {
+ if (!$this->persist_) {
+ @fclose($this->handle_);
+ $this->handle_ = null;
+ }
+ }
+
+ /**
+ * Uses stream get contents to do the reading
+ *
+ * @param int $len How many bytes
+ * @return string Binary data
+ */
+ public function readAll($len) {
+ if ($this->sendTimeoutSet_) {
+ stream_set_timeout($this->handle_, 0, $this->recvTimeout_*1000);
+ $this->sendTimeoutSet_ = FALSE;
+ }
+ // This call does not obey stream_set_timeout values!
+ // $buf = @stream_get_contents($this->handle_, $len);
+
+ $pre = null;
+ while (TRUE) {
+ $buf = @fread($this->handle_, $len);
+ if ($buf === FALSE || $buf === '') {
+ $md = stream_get_meta_data($this->handle_);
+ if ($md['timed_out']) {
+ throw new TException('TSocket: timed out reading '.$len.' bytes from '.
+ $this->host_.':'.$this->port_);
+ } else {
+ throw new TException('TSocket: Could not read '.$len.' bytes from '.
+ $this->host_.':'.$this->port_);
+ }
+ } else if (($sz = strlen($buf)) < $len) {
+ $md = stream_get_meta_data($this->handle_);
+ if ($md['timed_out']) {
+ throw new TException('TSocket: timed out reading '.$len.' bytes from '.
+ $this->host_.':'.$this->port_);
+ } else {
+ $pre .= $buf;
+ $len -= $sz;
+ }
+ } else {
+ return $pre.$buf;
+ }
+ }
+ }
+
+ /**
+ * Read from the socket
+ *
+ * @param int $len How many bytes
+ * @return string Binary data
+ */
+ public function read($len) {
+ if ($this->sendTimeoutSet_) {
+ stream_set_timeout($this->handle_, 0, $this->recvTimeout_*1000);
+ $this->sendTimeoutSet_ = FALSE;
+ }
+ $data = @fread($this->handle_, $len);
+ if ($data === FALSE || $data === '') {
+ $md = stream_get_meta_data($this->handle_);
+ if ($md['timed_out']) {
+ throw new TException('TSocket: timed out reading '.$len.' bytes from '.
+ $this->host_.':'.$this->port_);
+ } else {
+ throw new TException('TSocket: Could not read '.$len.' bytes from '.
+ $this->host_.':'.$this->port_);
+ }
+ }
+ return $data;
+ }
+
+ /**
+ * Write to the socket.
+ *
+ * @param string $buf The data to write
+ */
+ public function write($buf) {
+ if (!$this->sendTimeoutSet_) {
+ stream_set_timeout($this->handle_, 0, $this->sendTimeout_*1000);
+ $this->sendTimeoutSet_ = TRUE;
+ }
+ while (strlen($buf) > 0) {
+ $got = @fwrite($this->handle_, $buf);
+ if ($got === 0 || $got === FALSE) {
+ $md = stream_get_meta_data($this->handle_);
+ if ($md['timed_out']) {
+ throw new TException('TSocket: timed out writing '.strlen($buf).' bytes from '.
+ $this->host_.':'.$this->port_);
+ } else {
+ throw new TException('TSocket: Could not write '.strlen($buf).' bytes '.
+ $this->host_.':'.$this->port_);
+ }
+ }
+ $buf = substr($buf, $got);
+ }
+ }
+
+ /**
+ * Flush output to the socket.
+ */
+ public function flush() {
+ $ret = fflush($this->handle_);
+ if ($ret === FALSE) {
+ throw new TException('TSocket: Could not flush: '.
+ $this->host_.':'.$this->port_);
+ }
+ }
+}
+
+?>
diff --git a/lib/php/src/transport/TSocketPool.php b/lib/php/src/transport/TSocketPool.php
new file mode 100644
index 0000000..7f1157c
--- /dev/null
+++ b/lib/php/src/transport/TSocketPool.php
@@ -0,0 +1,296 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/** Inherits from Socket */
+include_once $GLOBALS['THRIFT_ROOT'].'/transport/TSocket.php';
+
+/**
+ * This library makes use of APC cache to make hosts as down in a web
+ * environment. If you are running from the CLI or on a system without APC
+ * installed, then these null functions will step in and act like cache
+ * misses.
+ */
+if (!function_exists('apc_fetch')) {
+ function apc_fetch($key) { return FALSE; }
+ function apc_store($key, $var, $ttl=0) { return FALSE; }
+}
+
+/**
+ * Sockets implementation of the TTransport interface that allows connection
+ * to a pool of servers.
+ *
+ * @package thrift.transport
+ */
+class TSocketPool extends TSocket {
+
+ /**
+ * Remote servers. Array of associative arrays with 'host' and 'port' keys
+ */
+ private $servers_ = array();
+
+ /**
+ * How many times to retry each host in connect
+ *
+ * @var int
+ */
+ private $numRetries_ = 1;
+
+ /**
+ * Retry interval in seconds, how long to not try a host if it has been
+ * marked as down.
+ *
+ * @var int
+ */
+ private $retryInterval_ = 60;
+
+ /**
+ * Max consecutive failures before marking a host down.
+ *
+ * @var int
+ */
+ private $maxConsecutiveFailures_ = 1;
+
+ /**
+ * Try hosts in order? or Randomized?
+ *
+ * @var bool
+ */
+ private $randomize_ = TRUE;
+
+ /**
+ * Always try last host, even if marked down?
+ *
+ * @var bool
+ */
+ private $alwaysTryLast_ = TRUE;
+
+ /**
+ * Socket pool constructor
+ *
+ * @param array $hosts List of remote hostnames
+ * @param mixed $ports Array of remote ports, or a single common port
+ * @param bool $persist Whether to use a persistent socket
+ * @param mixed $debugHandler Function for error logging
+ */
+ public function __construct($hosts=array('localhost'),
+ $ports=array(9090),
+ $persist=FALSE,
+ $debugHandler=null) {
+ parent::__construct(null, 0, $persist, $debugHandler);
+
+ if (!is_array($ports)) {
+ $port = $ports;
+ $ports = array();
+ foreach ($hosts as $key => $val) {
+ $ports[$key] = $port;
+ }
+ }
+
+ foreach ($hosts as $key => $host) {
+ $this->servers_ []= array('host' => $host,
+ 'port' => $ports[$key]);
+ }
+ }
+
+ /**
+ * Add a server to the pool
+ *
+ * This function does not prevent you from adding a duplicate server entry.
+ *
+ * @param string $host hostname or IP
+ * @param int $port port
+ */
+ public function addServer($host, $port) {
+ $this->servers_[] = array('host' => $host, 'port' => $port);
+ }
+
+ /**
+ * Sets how many time to keep retrying a host in the connect function.
+ *
+ * @param int $numRetries
+ */
+ public function setNumRetries($numRetries) {
+ $this->numRetries_ = $numRetries;
+ }
+
+ /**
+ * Sets how long to wait until retrying a host if it was marked down
+ *
+ * @param int $numRetries
+ */
+ public function setRetryInterval($retryInterval) {
+ $this->retryInterval_ = $retryInterval;
+ }
+
+ /**
+ * Sets how many time to keep retrying a host before marking it as down.
+ *
+ * @param int $numRetries
+ */
+ public function setMaxConsecutiveFailures($maxConsecutiveFailures) {
+ $this->maxConsecutiveFailures_ = $maxConsecutiveFailures;
+ }
+
+ /**
+ * Turns randomization in connect order on or off.
+ *
+ * @param bool $randomize
+ */
+ public function setRandomize($randomize) {
+ $this->randomize_ = $randomize;
+ }
+
+ /**
+ * Whether to always try the last server.
+ *
+ * @param bool $alwaysTryLast
+ */
+ public function setAlwaysTryLast($alwaysTryLast) {
+ $this->alwaysTryLast_ = $alwaysTryLast;
+ }
+
+
+ /**
+ * Connects the socket by iterating through all the servers in the pool
+ * and trying to find one that works.
+ */
+ public function open() {
+ // Check if we want order randomization
+ if ($this->randomize_) {
+ shuffle($this->servers_);
+ }
+
+ // Count servers to identify the "last" one
+ $numServers = count($this->servers_);
+
+ for ($i = 0; $i < $numServers; ++$i) {
+
+ // This extracts the $host and $port variables
+ extract($this->servers_[$i]);
+
+ // Check APC cache for a record of this server being down
+ $failtimeKey = 'thrift_failtime:'.$host.':'.$port.'~';
+
+ // Cache miss? Assume it's OK
+ $lastFailtime = apc_fetch($failtimeKey);
+ if ($lastFailtime === FALSE) {
+ $lastFailtime = 0;
+ }
+
+ $retryIntervalPassed = FALSE;
+
+ // Cache hit...make sure enough the retry interval has elapsed
+ if ($lastFailtime > 0) {
+ $elapsed = time() - $lastFailtime;
+ if ($elapsed > $this->retryInterval_) {
+ $retryIntervalPassed = TRUE;
+ if ($this->debug_) {
+ call_user_func($this->debugHandler_,
+ 'TSocketPool: retryInterval '.
+ '('.$this->retryInterval_.') '.
+ 'has passed for host '.$host.':'.$port);
+ }
+ }
+ }
+
+ // Only connect if not in the middle of a fail interval, OR if this
+ // is the LAST server we are trying, just hammer away on it
+ $isLastServer = FALSE;
+ if ($this->alwaysTryLast_) {
+ $isLastServer = ($i == ($numServers - 1));
+ }
+
+ if (($lastFailtime === 0) ||
+ ($isLastServer) ||
+ ($lastFailtime > 0 && $retryIntervalPassed)) {
+
+ // Set underlying TSocket params to this one
+ $this->host_ = $host;
+ $this->port_ = $port;
+
+ // Try up to numRetries_ connections per server
+ for ($attempt = 0; $attempt < $this->numRetries_; $attempt++) {
+ try {
+ // Use the underlying TSocket open function
+ parent::open();
+
+ // Only clear the failure counts if required to do so
+ if ($lastFailtime > 0) {
+ apc_store($failtimeKey, 0);
+ }
+
+ // Successful connection, return now
+ return;
+
+ } catch (TException $tx) {
+ // Connection failed
+ }
+ }
+
+ // Mark failure of this host in the cache
+ $consecfailsKey = 'thrift_consecfails:'.$host.':'.$port.'~';
+
+ // Ignore cache misses
+ $consecfails = apc_fetch($consecfailsKey);
+ if ($consecfails === FALSE) {
+ $consecfails = 0;
+ }
+
+ // Increment by one
+ $consecfails++;
+
+ // Log and cache this failure
+ if ($consecfails >= $this->maxConsecutiveFailures_) {
+ if ($this->debug_) {
+ call_user_func($this->debugHandler_,
+ 'TSocketPool: marking '.$host.':'.$port.
+ ' as down for '.$this->retryInterval_.' secs '.
+ 'after '.$consecfails.' failed attempts.');
+ }
+ // Store the failure time
+ apc_store($failtimeKey, time());
+
+ // Clear the count of consecutive failures
+ apc_store($consecfailsKey, 0);
+ } else {
+ apc_store($consecfailsKey, $consecfails);
+ }
+ }
+ }
+
+ // Holy shit we failed them all. The system is totally ill!
+ $error = 'TSocketPool: All hosts in pool are down. ';
+ $hosts = array();
+ foreach ($this->servers_ as $server) {
+ $hosts []= $server['host'].':'.$server['port'];
+ }
+ $hostlist = implode(',', $hosts);
+ $error .= '('.$hostlist.')';
+ if ($this->debug_) {
+ call_user_func($this->debugHandler_, $error);
+ }
+ throw new TException($error);
+ }
+}
+
+?>
diff --git a/lib/php/src/transport/TTransport.php b/lib/php/src/transport/TTransport.php
new file mode 100644
index 0000000..e244525
--- /dev/null
+++ b/lib/php/src/transport/TTransport.php
@@ -0,0 +1,108 @@
+<?php
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * @package thrift.transport
+ */
+
+
+/**
+ * Transport exceptions
+ */
+class TTransportException extends TException {
+
+ const UNKNOWN = 0;
+ const NOT_OPEN = 1;
+ const ALREADY_OPEN = 2;
+ const TIMED_OUT = 3;
+ const END_OF_FILE = 4;
+
+ function __construct($message=null, $code=0) {
+ parent::__construct($message, $code);
+ }
+}
+
+/**
+ * Base interface for a transport agent.
+ *
+ * @package thrift.transport
+ */
+abstract class TTransport {
+
+ /**
+ * Whether this transport is open.
+ *
+ * @return boolean true if open
+ */
+ public abstract function isOpen();
+
+ /**
+ * Open the transport for reading/writing
+ *
+ * @throws TTransportException if cannot open
+ */
+ public abstract function open();
+
+ /**
+ * Close the transport.
+ */
+ public abstract function close();
+
+ /**
+ * Read some data into the array.
+ *
+ * @param int $len How much to read
+ * @return string The data that has been read
+ * @throws TTransportException if cannot read any more data
+ */
+ public abstract function read($len);
+
+ /**
+ * Guarantees that the full amount of data is read.
+ *
+ * @return string The data, of exact length
+ * @throws TTransportException if cannot read data
+ */
+ public function readAll($len) {
+ // return $this->read($len);
+
+ $data = '';
+ $got = 0;
+ while (($got = strlen($data)) < $len) {
+ $data .= $this->read($len - $got);
+ }
+ return $data;
+ }
+
+ /**
+ * Writes the given data out.
+ *
+ * @param string $buf The data to write
+ * @throws TTransportException if writing fails
+ */
+ public abstract function write($buf);
+
+ /**
+ * Flushes any pending data out of a buffer
+ *
+ * @throws TTransportException if a writing error occurs
+ */
+ public function flush() {}
+}
+
+?>