def _try_read_headers()

in ossdbtoolsservice/hosting/json_reader.py [0:0]

• 33 lines of code
• 11 McCabe index (conditional complexity)

    def _try_read_headers(self):
        """
        Try to read the header information from the internal buffer expecting the last header to contain '\r\n\r\n'
        :raises LookupError: The content-length header was not found
        :raises ValueError: The content-length contained an invalid literal for int
        :raises KeyError: The header block was malformed by not having a key:value format
        :return: True on successful read of headers, False on failure to find headers
        """
        # Scan the buffer up until right before the \r\n\r\n
        scan_offset = self._read_offset
        while scan_offset + 3 < self._buffer_end_offset and (
            self._buffer[scan_offset] != self.CR or
            self._buffer[scan_offset + 1] != self.LF or
            self._buffer[scan_offset + 2] != self.CR or
            self._buffer[scan_offset + 3] != self.LF
        ):
            scan_offset += 1

        # If we reached the end of the buffer and haven't found the control sequence, we haven't found the headers
        if scan_offset + 3 >= self._buffer_end_offset:
            return False

        # Split the headers by newline
        try:
            headers_read = self._buffer[self._read_offset:scan_offset].decode('ascii')
            for header in headers_read.split('\n'):
                colon_index = header.find(':')

                # Make sure there's a colon to split key and value on
                if colon_index == -1:
                    if self._logger is not None:
                        self._logger.warn('JSON RPC reader encountered missing colons in try_read_headers()')
                    raise KeyError('Colon missing from header: {}'.format(header))

                # Case insensitive check
                header_key = header[:colon_index].strip().lower()
                header_value = header[colon_index + 1:].strip()
                self._headers[header_key] = header_value

            # Was content-length found?
            if 'content-length' not in self._headers:
                if self._logger is not None:
                    self._logger.warn('JSON RPC reader did not find Content-Length in the headers')
                raise LookupError('Content-Length was not found in headers received.')

            self._expected_content_length = int(self._headers['content-length'])

        except (ValueError, KeyError, LookupError):
            # ValueError: Content-Length contained invalid literal for int
            # KeyError: Headers were malformed due to missing colon
            # LookupError: Content-Length header was not found
            self._trim_buffer_and_resize(scan_offset + 4)
            raise

        # Pushing read pointer past the newline character
        self._read_offset = scan_offset + 4
        self._read_state = self.ReadState.Content

        return True