[ Index ]

MailPress 544

[ Index ]     [ Classes ]     [ Functions ]     [ Variables ]     [ Constants ]     [ Statistics ]    

title

Body

[close]

/mp-includes/Swiftmailer/classes/Swift/Transport/ -> StreamBuffer.php (source)

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


Generated: Thu Apr 28 18:38:52 2016 Cross-referenced by PHPXref 0.7.1