vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Message.php

   1 <?php
   2 /* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
   3 // +---------------------------------------------------------------------------
   4 // | SWAN [ $_SWANBR_SLOGAN_$ ]
   5 // +---------------------------------------------------------------------------
   6 // | Copyright $_SWANBR_COPYRIGHT_$
   7 // +---------------------------------------------------------------------------
   8 // | Version  $_SWANBR_VERSION_$
   9 // +---------------------------------------------------------------------------
  10 // | Licensed ( $_SWANBR_LICENSED_URL_$ )
  11 // +---------------------------------------------------------------------------
  12 // | $_SWANBR_WEB_DOMAIN_$
  13 // +---------------------------------------------------------------------------
  14
  15 namespace Kafka\Protocol\Fetch;
  16
  17 use \Kafka\Protocol\Decoder;
  18
  19 /**
  20 +------------------------------------------------------------------------------
  21 * Kafka protocol since Kafka v0.8
  22 +------------------------------------------------------------------------------
  23 *
  24 * @package
  25 * @version $_SWANBR_VERSION_$
  26 * @copyright Copyleft
  27 * @author $_SWANBR_AUTHOR_$
  28 +------------------------------------------------------------------------------
  29 */
  30
  31 class Message
  32 {
  33     // {{{ members
  34
  35     /**
  36      * crc32 code
  37      *
  38      * @var float
  39      * @access private
  40      */
  41     private $crc = 0;
  42
  43     /**
  44      * This is a version id used to allow backwards compatible evolution of the
  45      * message binary format.
  46      *
  47      * @var float
  48      * @access private
  49      */
  50     private $magic = 0;
  51
  52     /**
  53      * The lowest 2 bits contain the compression codec used for the message. The
  54      * other bits should be set to 0.
  55      *
  56      * @var float
  57      * @access private
  58      */
  59     private $attribute = 0;
  60
  61     /**
  62      * message key
  63      *
  64      * @var string
  65      * @access private
  66      */
  67     private $key = '';
  68
  69     /**
  70      * message value
  71      *
  72      * @var string
  73      * @access private
  74      */
  75     private $value = '';
  76
  77     // }}}
  78     // {{{ functions
  79     // {{{ public function __construct()
  80
  81     /**
  82      * __construct
  83      *
  84      * @param string(raw) $msg
  85      * @access public
  86      */
  87     public function __construct($msg)
  88     {
  89         $offset = 0;
  90         $crc = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
  91         $offset += 4;
  92         $this->crc = array_shift($crc);
  93         $magic = Decoder::unpack(Decoder::BIT_B8, substr($msg, $offset, 1));
  94         $this->magic = array_shift($magic);
  95         $offset += 1;
  96         $attr  = Decoder::unpack(Decoder::BIT_B8, substr($msg, $offset, 1));
  97         $this->attribute = array_shift($attr);
  98         $offset += 1;
  99         $keyLen = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
 100         $keyLen = array_shift($keyLen);
 101         $offset += 4;
 102         if ($keyLen > 0 && $keyLen != 0xFFFFFFFF) {
 103             $this->key = substr($msg, $offset, $keyLen);
 104             $offset += $keyLen;
 105         }
 106         $messageSize = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
 107         $messageSize = array_shift($messageSize);
 108         $offset += 4;
 109         if ($messageSize) {
 110             $this->value = substr($msg, $offset, $messageSize);
 111         }
 112     }
 113
 114     // }}}
 115     // {{{ public function getMessage()
 116
 117     /**
 118      * get message data
 119      *
 120      * @access public
 121      * @return string (raw)
 122      */
 123     public function getMessage()
 124     {
 125         return $this->value;
 126     }
 127
 128     // }}}
 129     // {{{ public function getMessageKey()
 130
 131     /**
 132      * get message key
 133      *
 134      * @access public
 135      * @return string (raw)
 136      */
 137     public function getMessageKey()
 138     {
 139         return $this->key;
 140     }
 141
 142     // }}}
 143     // {{{ public function __toString()
 144
 145     /**
 146      * __toString
 147      *
 148      * @access public
 149      * @return string (raw)
 150      */
 151     public function __toString()
 152     {
 153         return $this->value;
 154     }
 155
 156     // }}}
 157     // }}}
 158 }