Decode a 'RecordIO' formatted message to its original type. :param data: an array of 'UTF-8' encoded bytes that make up a partial 'RecordIO' message. Subsequent calls to this function maintain state to build up a full 'RecordIO'
(self, data)
| 118 | self.length = 0 |
| 119 | |
| 120 | def decode(self, data): |
| 121 | """ |
| 122 | Decode a 'RecordIO' formatted message to its original type. |
| 123 | |
| 124 | :param data: an array of 'UTF-8' encoded bytes that make up a |
| 125 | partial 'RecordIO' message. Subsequent calls to this |
| 126 | function maintain state to build up a full 'RecordIO' |
| 127 | message and decode it |
| 128 | :type data: bytes |
| 129 | :returns: a list of deserialized messages |
| 130 | :rtype: list |
| 131 | """ |
| 132 | |
| 133 | if not isinstance(data, bytes): |
| 134 | raise MesosException("Parameter 'data' must of of type 'bytes'") |
| 135 | |
| 136 | if self.state == self.FAILED: |
| 137 | raise MesosException("Decoder is in a FAILED state") |
| 138 | |
| 139 | records = [] |
| 140 | |
| 141 | for c in data: |
| 142 | if self.state == self.HEADER: |
| 143 | if c != ord('\n'): |
| 144 | self.buffer += bytes([c]) |
| 145 | continue |
| 146 | |
| 147 | try: |
| 148 | self.length = int(self.buffer.decode("UTF-8")) |
| 149 | except Exception as exception: |
| 150 | self.state = self.FAILED |
| 151 | raise MesosException("Failed to decode length" |
| 152 | "'{buffer}': {error}" |
| 153 | .format(buffer=self.buffer, |
| 154 | error=exception)) |
| 155 | |
| 156 | self.buffer = bytes("", "UTF-8") |
| 157 | self.state = self.RECORD |
| 158 | |
| 159 | # Note that for 0 length records, we immediately decode. |
| 160 | if self.length <= 0: |
| 161 | records.append(self.deserialize(self.buffer)) |
| 162 | self.state = self.HEADER |
| 163 | |
| 164 | elif self.state == self.RECORD: |
| 165 | assert self.length |
| 166 | assert len(self.buffer) < self.length |
| 167 | |
| 168 | self.buffer += bytes([c]) |
| 169 | |
| 170 | if len(self.buffer) == self.length: |
| 171 | records.append(self.deserialize(self.buffer)) |
| 172 | self.buffer = bytes("", "UTF-8") |
| 173 | self.state = self.HEADER |
| 174 | |
| 175 | return records |