Lines Matching refs:channel_id
99 channel_id = ord(data[offset])
101 if channel_id & 0xe0 == 0xe0:
104 channel_id = struct.unpack('!L',
107 elif channel_id & 0xc0 == 0xc0:
110 channel_id = (((channel_id & 0x1f) << 16) +
113 elif channel_id & 0x80 == 0x80:
116 channel_id = struct.unpack('!H', data[offset:offset+2])[0] & 0x3fff
119 return channel_id, channel_id_length
167 channel_id, advance = _parse_channel_id(data, pos)
168 block.channel_id = channel_id
180 channel_id, channel_id_length = _parse_channel_id(data, pos)
181 block.channel_id = channel_id
191 channel_id, advance = _parse_channel_id(data, pos)
192 block.channel_id = channel_id
213 def _encode_channel_id(channel_id):
214 if channel_id < 0:
215 raise ValueError('Channel id %d must not be negative' % channel_id)
217 if channel_id < 2 ** 7:
218 return chr(channel_id)
219 if channel_id < 2 ** 14:
220 return struct.pack('!H', 0x8000 + channel_id)
221 if channel_id < 2 ** 21:
222 first = chr(0xc0 + (channel_id >> 16))
223 return first + struct.pack('!H', channel_id & 0xffff)
224 if channel_id < 2 ** 29:
225 return struct.pack('!L', 0xe0000000 + channel_id)
227 raise ValueError('Channel id %d is too large' % channel_id)
258 def _create_add_channel_request(channel_id, encoded_handshake,
267 return (chr(first_byte) + _encode_channel_id(channel_id) +
271 def _create_flow_control(channel_id, replenished_quota):
274 return (chr(first_byte) + _encode_channel_id(channel_id) +
347 channel_id, advance = _parse_channel_id(message)
348 self._mux._dispatch_frame(channel_id, message[advance:])
428 if not block.channel_id in self._logical_channels:
430 'channel id %d' % block.channel_id)
431 self._logical_channels[block.channel_id].send_quota += (
439 def _dispatch_frame(self, channel_id, payload):
440 if channel_id == _CONTROL_CHANNEL_ID:
451 if not channel_id in self._logical_channels:
454 channel_id)
457 self._logical_channels[channel_id].receive_quota -= (
459 if self._logical_channels[channel_id].receive_quota < 0:
461 channel_id)
464 self._logical_channels[channel_id].queue.put(inner_frame)
494 def _assert_send_quota_available(self, channel_id):
497 if self._logical_channels[channel_id].send_quota == 0:
503 if self._logical_channels[channel_id].send_quota == 0:
505 channel_id)
531 def add_channel(self, channel_id, options):
535 if channel_id in self._logical_channels:
536 raise Exception('Channel id %d already exists' % channel_id)
567 channel_id, handshake)
580 if block.channel_id == channel_id:
630 self._logical_channels[channel_id] = _LogicalChannelData()
631 self._logical_channels[channel_id].send_quota = send_quota
634 self._logger.debug('Logical channel %d established' % channel_id)
636 def _check_logical_channel_is_opened(self, channel_id):
640 if not channel_id in self._logical_channels:
643 def drop_channel(self, channel_id):
647 def send_flow_control(self, channel_id, replenished_quota):
648 self._check_logical_channel_is_opened(channel_id)
649 flow_control = _create_flow_control(channel_id, replenished_quota)
654 self._logical_channels[channel_id].receive_quota += (
660 def send_message(self, channel_id, message, end=True, binary=False):
661 self._check_logical_channel_is_opened(channel_id)
671 if self._logical_channels[channel_id].send_quota < len(message):
673 self._logical_channels[channel_id].send_quota,
676 self._logical_channels[channel_id].send_quota -= len(message)
679 payload = _encode_channel_id(channel_id) + chr(first_byte) + message
682 def assert_receive(self, channel_id, payload, binary=False):
683 self._check_logical_channel_is_opened(channel_id)
686 inner_frame = self._logical_channels[channel_id].queue.get(
690 channel_id)
704 def send_close(self, channel_id, code=None, reason=''):
705 self._check_logical_channel_is_opened(channel_id)
713 payload = _encode_channel_id(channel_id) + chr(first_byte) + body
716 def assert_receive_close(self, channel_id):
717 self._check_logical_channel_is_opened(channel_id)
720 inner_frame = self._logical_channels[channel_id].queue.get(
724 channel_id)
