StreamBuffer.php 9.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316
  1. <?php
  2. /*
  3. * This file is part of SwiftMailer.
  4. * (c) 2004-2009 Chris Corbyn
  5. *
  6. * For the full copyright and license information, please view the LICENSE
  7. * file that was distributed with this source code.
  8. */
  9. /**
  10. * A generic IoBuffer implementation supporting remote sockets and local processes.
  11. *
  12. * @package Swift
  13. * @subpackage Transport
  14. * @author Chris Corbyn
  15. */
  16. class Swift_Transport_StreamBuffer extends Swift_ByteStream_AbstractFilterableInputStream implements Swift_Transport_IoBuffer
  17. {
  18. /** A primary socket */
  19. private $_stream;
  20. /** The input stream */
  21. private $_in;
  22. /** The output stream */
  23. private $_out;
  24. /** Buffer initialization parameters */
  25. private $_params = array();
  26. /** The ReplacementFilterFactory */
  27. private $_replacementFactory;
  28. /** Translations performed on data being streamed into the buffer */
  29. private $_translations = array();
  30. /**
  31. * Create a new StreamBuffer using $replacementFactory for transformations.
  32. *
  33. * @param Swift_ReplacementFilterFactory $replacementFactory
  34. */
  35. public function __construct(Swift_ReplacementFilterFactory $replacementFactory)
  36. {
  37. $this->_replacementFactory = $replacementFactory;
  38. }
  39. /**
  40. * Perform any initialization needed, using the given $params.
  41. *
  42. * Parameters will vary depending upon the type of IoBuffer used.
  43. *
  44. * @param array $params
  45. */
  46. public function initialize(array $params)
  47. {
  48. $this->_params = $params;
  49. switch ($params['type']) {
  50. case self::TYPE_PROCESS:
  51. $this->_establishProcessConnection();
  52. break;
  53. case self::TYPE_SOCKET:
  54. default:
  55. $this->_establishSocketConnection();
  56. break;
  57. }
  58. }
  59. /**
  60. * Set an individual param on the buffer (e.g. switching to SSL).
  61. *
  62. * @param string $param
  63. * @param mixed $value
  64. */
  65. public function setParam($param, $value)
  66. {
  67. if (isset($this->_stream)) {
  68. switch ($param) {
  69. case 'timeout':
  70. if ($this->_stream) {
  71. stream_set_timeout($this->_stream, $value);
  72. }
  73. break;
  74. case 'blocking':
  75. if ($this->_stream) {
  76. stream_set_blocking($this->_stream, 1);
  77. }
  78. }
  79. }
  80. $this->_params[$param] = $value;
  81. }
  82. public function startTLS()
  83. {
  84. return stream_socket_enable_crypto($this->_stream, true, STREAM_CRYPTO_METHOD_TLS_CLIENT);
  85. }
  86. /**
  87. * Perform any shutdown logic needed.
  88. */
  89. public function terminate()
  90. {
  91. if (isset($this->_stream)) {
  92. switch ($this->_params['type']) {
  93. case self::TYPE_PROCESS:
  94. fclose($this->_in);
  95. fclose($this->_out);
  96. proc_close($this->_stream);
  97. break;
  98. case self::TYPE_SOCKET:
  99. default:
  100. fclose($this->_stream);
  101. break;
  102. }
  103. }
  104. $this->_stream = null;
  105. $this->_out = null;
  106. $this->_in = null;
  107. }
  108. /**
  109. * Set an array of string replacements which should be made on data written
  110. * to the buffer.
  111. *
  112. * This could replace LF with CRLF for example.
  113. *
  114. * @param string[] $replacements
  115. */
  116. public function setWriteTranslations(array $replacements)
  117. {
  118. foreach ($this->_translations as $search => $replace) {
  119. if (!isset($replacements[$search])) {
  120. $this->removeFilter($search);
  121. unset($this->_translations[$search]);
  122. }
  123. }
  124. foreach ($replacements as $search => $replace) {
  125. if (!isset($this->_translations[$search])) {
  126. $this->addFilter(
  127. $this->_replacementFactory->createFilter($search, $replace), $search
  128. );
  129. $this->_translations[$search] = true;
  130. }
  131. }
  132. }
  133. /**
  134. * Get a line of output (including any CRLF).
  135. *
  136. * The $sequence number comes from any writes and may or may not be used
  137. * depending upon the implementation.
  138. *
  139. * @param integer $sequence of last write to scan from
  140. *
  141. * @return string
  142. *
  143. * @throws Swift_IoException
  144. */
  145. public function readLine($sequence)
  146. {
  147. if (isset($this->_out) && !feof($this->_out)) {
  148. $line = fgets($this->_out);
  149. if (strlen($line)==0) {
  150. $metas = stream_get_meta_data($this->_out);
  151. if ($metas['timed_out']) {
  152. throw new Swift_IoException(
  153. 'Connection to ' .
  154. $this->_getReadConnectionDescription() .
  155. ' Timed Out'
  156. );
  157. }
  158. }
  159. return $line;
  160. }
  161. }
  162. /**
  163. * Reads $length bytes from the stream into a string and moves the pointer
  164. * through the stream by $length.
  165. *
  166. * If less bytes exist than are requested the remaining bytes are given instead.
  167. * If no bytes are remaining at all, boolean false is returned.
  168. *
  169. * @param integer $length
  170. *
  171. * @return string|boolean
  172. *
  173. * @throws Swift_IoException
  174. */
  175. public function read($length)
  176. {
  177. if (isset($this->_out) && !feof($this->_out)) {
  178. $ret = fread($this->_out, $length);
  179. if (strlen($ret)==0) {
  180. $metas = stream_get_meta_data($this->_out);
  181. if ($metas['timed_out']) {
  182. throw new Swift_IoException(
  183. 'Connection to ' .
  184. $this->_getReadConnectionDescription() .
  185. ' Timed Out'
  186. );
  187. }
  188. }
  189. return $ret;
  190. }
  191. }
  192. /** Not implemented */
  193. public function setReadPointer($byteOffset)
  194. {
  195. }
  196. // -- Protected methods
  197. /** Flush the stream contents */
  198. protected function _flush()
  199. {
  200. if (isset($this->_in)) {
  201. fflush($this->_in);
  202. }
  203. }
  204. /** Write this bytes to the stream */
  205. protected function _commit($bytes)
  206. {
  207. if (isset($this->_in)
  208. && fwrite($this->_in, $bytes))
  209. {
  210. return ++$this->_sequence;
  211. }
  212. }
  213. // -- Private methods
  214. /**
  215. * Establishes a connection to a remote server.
  216. */
  217. private function _establishSocketConnection()
  218. {
  219. $host = $this->_params['host'];
  220. if (!empty($this->_params['protocol'])) {
  221. $host = $this->_params['protocol'] . '://' . $host;
  222. }
  223. $timeout = 15;
  224. if (!empty($this->_params['timeout'])) {
  225. $timeout = $this->_params['timeout'];
  226. }
  227. $options = array();
  228. if (!empty($this->_params['sourceIp'])) {
  229. $options['socket']['bindto']=$this->_params['sourceIp'].':0';
  230. }
  231. $this->_stream = @stream_socket_client($host.':'.$this->_params['port'], $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, stream_context_create($options));
  232. if (false === $this->_stream) {
  233. throw new Swift_TransportException(
  234. 'Connection could not be established with host ' . $this->_params['host'] .
  235. ' [' . $errstr . ' #' . $errno . ']'
  236. );
  237. }
  238. if (!empty($this->_params['blocking'])) {
  239. stream_set_blocking($this->_stream, 1);
  240. } else {
  241. stream_set_blocking($this->_stream, 0);
  242. }
  243. stream_set_timeout($this->_stream, $timeout);
  244. $this->_in =& $this->_stream;
  245. $this->_out =& $this->_stream;
  246. }
  247. /**
  248. * Opens a process for input/output.
  249. */
  250. private function _establishProcessConnection()
  251. {
  252. $command = $this->_params['command'];
  253. $descriptorSpec = array(
  254. 0 => array('pipe', 'r'),
  255. 1 => array('pipe', 'w'),
  256. 2 => array('pipe', 'w')
  257. );
  258. $this->_stream = proc_open($command, $descriptorSpec, $pipes);
  259. stream_set_blocking($pipes[2], 0);
  260. if ($err = stream_get_contents($pipes[2])) {
  261. throw new Swift_TransportException(
  262. 'Process could not be started [' . $err . ']'
  263. );
  264. }
  265. $this->_in =& $pipes[0];
  266. $this->_out =& $pipes[1];
  267. }
  268. private function _getReadConnectionDescription()
  269. {
  270. switch ($this->_params['type']) {
  271. case self::TYPE_PROCESS:
  272. return 'Process '.$this->_params['command'];
  273. break;
  274. case self::TYPE_SOCKET:
  275. default:
  276. $host = $this->_params['host'];
  277. if (!empty($this->_params['protocol'])) {
  278. $host = $this->_params['protocol'] . '://' . $host;
  279. }
  280. $host.=':'.$this->_params['port'];
  281. return $host;
  282. break;
  283. }
  284. }
  285. }