| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198 | # Copyright 2016 gRPC authors.## Licensed under the Apache License, Version 2.0 (the "License");# you may not use this file except in compliance with the License.# You may obtain a copy of the License at##     http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.import loggingimport messages_pb2import structimport h2import h2.connectionimport twistedimport twisted.internetimport twisted.internet.protocol_READ_CHUNK_SIZE = 16384_GRPC_HEADER_SIZE = 5_MIN_SETTINGS_MAX_FRAME_SIZE = 16384class H2ProtocolBaseServer(twisted.internet.protocol.Protocol):  def __init__(self):    self._conn = h2.connection.H2Connection(client_side=False)    self._recv_buffer = {}    self._handlers = {}    self._handlers['ConnectionMade'] = self.on_connection_made_default    self._handlers['DataReceived'] = self.on_data_received_default    self._handlers['WindowUpdated'] = self.on_window_update_default    self._handlers['RequestReceived'] = self.on_request_received_default    self._handlers['SendDone'] = self.on_send_done_default    self._handlers['ConnectionLost'] = self.on_connection_lost    self._handlers['PingAcknowledged'] = self.on_ping_acknowledged_default    self._stream_status = {}    self._send_remaining = {}    self._outstanding_pings = 0  def set_handlers(self, handlers):    self._handlers = handlers  def connectionMade(self):    self._handlers['ConnectionMade']()  def connectionLost(self, reason):    self._handlers['ConnectionLost'](reason)  def on_connection_made_default(self):    logging.info('Connection Made')    self._conn.initiate_connection()    self.transport.setTcpNoDelay(True)    self.transport.write(self._conn.data_to_send())  def on_connection_lost(self, reason):    logging.info('Disconnected %s' % reason)  def dataReceived(self, data):    try:      events = self._conn.receive_data(data)    except h2.exceptions.ProtocolError:      # this try/except block catches exceptions due to race between sending      # GOAWAY and processing a response in flight.      return    if self._conn.data_to_send:      self.transport.write(self._conn.data_to_send())    for event in events:      if isinstance(event, h2.events.RequestReceived) and self._handlers.has_key('RequestReceived'):        logging.info('RequestReceived Event for stream: %d' % event.stream_id)        self._handlers['RequestReceived'](event)      elif isinstance(event, h2.events.DataReceived) and self._handlers.has_key('DataReceived'):        logging.info('DataReceived Event for stream: %d' % event.stream_id)        self._handlers['DataReceived'](event)      elif isinstance(event, h2.events.WindowUpdated) and self._handlers.has_key('WindowUpdated'):        logging.info('WindowUpdated Event for stream: %d' % event.stream_id)        self._handlers['WindowUpdated'](event)      elif isinstance(event, h2.events.PingAcknowledged) and self._handlers.has_key('PingAcknowledged'):        logging.info('PingAcknowledged Event')        self._handlers['PingAcknowledged'](event)    self.transport.write(self._conn.data_to_send())  def on_ping_acknowledged_default(self, event):    logging.info('ping acknowledged')    self._outstanding_pings -= 1  def on_data_received_default(self, event):    self._conn.acknowledge_received_data(len(event.data), event.stream_id)    self._recv_buffer[event.stream_id] += event.data  def on_request_received_default(self, event):    self._recv_buffer[event.stream_id] = ''    self._stream_id = event.stream_id    self._stream_status[event.stream_id] = True    self._conn.send_headers(      stream_id=event.stream_id,      headers=[          (':status', '200'),          ('content-type', 'application/grpc'),          ('grpc-encoding', 'identity'),          ('grpc-accept-encoding', 'identity,deflate,gzip'),      ],    )    self.transport.write(self._conn.data_to_send())  def on_window_update_default(self, _, pad_length=None, read_chunk_size=_READ_CHUNK_SIZE):    # try to resume sending on all active streams (update might be for connection)    for stream_id in self._send_remaining:      self.default_send(stream_id, pad_length=pad_length, read_chunk_size=read_chunk_size)  def send_reset_stream(self):    self._conn.reset_stream(self._stream_id)    self.transport.write(self._conn.data_to_send())  def setup_send(self, data_to_send, stream_id, pad_length=None, read_chunk_size=_READ_CHUNK_SIZE):    logging.info('Setting up data to send for stream_id: %d' % stream_id)    self._send_remaining[stream_id] = len(data_to_send)    self._send_offset = 0    self._data_to_send = data_to_send    self.default_send(stream_id, pad_length=pad_length, read_chunk_size=read_chunk_size)  def default_send(self, stream_id, pad_length=None, read_chunk_size=_READ_CHUNK_SIZE):    if not self._send_remaining.has_key(stream_id):      # not setup to send data yet      return    while self._send_remaining[stream_id] > 0:      lfcw = self._conn.local_flow_control_window(stream_id)      padding_bytes = pad_length + 1 if pad_length is not None else 0      if lfcw - padding_bytes <= 0:        logging.info('Stream %d. lfcw: %d. padding bytes: %d. not enough quota yet' % (stream_id, lfcw, padding_bytes))        break      chunk_size = min(lfcw - padding_bytes, read_chunk_size)      bytes_to_send = min(chunk_size, self._send_remaining[stream_id])      logging.info('flow_control_window = %d. sending [%d:%d] stream_id %d. includes %d total padding bytes' %                    (lfcw, self._send_offset, self._send_offset + bytes_to_send + padding_bytes,                    stream_id, padding_bytes))      # The receiver might allow sending frames larger than the http2 minimum      # max frame size (16384), but this test should never send more than 16384      # for simplicity (which is always legal).      if bytes_to_send + padding_bytes > _MIN_SETTINGS_MAX_FRAME_SIZE:        raise ValueError("overload: sending %d" % (bytes_to_send + padding_bytes))      data = self._data_to_send[self._send_offset : self._send_offset + bytes_to_send]      try:        self._conn.send_data(stream_id, data, end_stream=False, pad_length=pad_length)      except h2.exceptions.ProtocolError:        logging.info('Stream %d is closed' % stream_id)        break      self._send_remaining[stream_id] -= bytes_to_send      self._send_offset += bytes_to_send      if self._send_remaining[stream_id] == 0:        self._handlers['SendDone'](stream_id)  def default_ping(self):    logging.info('sending ping')    self._outstanding_pings += 1    self._conn.ping(b'\x00'*8)    self.transport.write(self._conn.data_to_send())  def on_send_done_default(self, stream_id):    if self._stream_status[stream_id]:      self._stream_status[stream_id] = False      self.default_send_trailer(stream_id)    else:      logging.error('Stream %d is already closed' % stream_id)  def default_send_trailer(self, stream_id):    logging.info('Sending trailer for stream id %d' % stream_id)    self._conn.send_headers(stream_id,      headers=[ ('grpc-status', '0') ],      end_stream=True    )    self.transport.write(self._conn.data_to_send())  @staticmethod  def default_response_data(response_size):    sresp = messages_pb2.SimpleResponse()    sresp.payload.body = b'\x00'*response_size    serialized_resp_proto = sresp.SerializeToString()    response_data = b'\x00' + struct.pack('i', len(serialized_resp_proto))[::-1] + serialized_resp_proto    return response_data  def parse_received_data(self, stream_id):    """ returns a grpc framed string of bytes containing response proto of the size    asked in request """    recv_buffer = self._recv_buffer[stream_id]    grpc_msg_size = struct.unpack('i',recv_buffer[1:5][::-1])[0]    if len(recv_buffer) != _GRPC_HEADER_SIZE + grpc_msg_size:      return None    req_proto_str = recv_buffer[5:5+grpc_msg_size]    sr = messages_pb2.SimpleRequest()    sr.ParseFromString(req_proto_str)    logging.info('Parsed simple request for stream %d' % stream_id)    return sr
 |