← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/smart.py

Clean up SmartServerStreamMedium implementations, including removing unnecessary flushes.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/smart.py
277
277
class SmartServerRequestProtocolOne(SmartProtocolBase):
278
278
    """Server-side encoding and decoding logic for smart version 1."""
279
279
    
280
 
    def __init__(self, output_stream, backing_transport):
281
 
        self._out_stream = output_stream
 
280
    def __init__(self, backing_transport, write_func):
282
281
        self._backing_transport = backing_transport
283
282
        self.excess_buffer = ''
284
283
        self._finished_reading = False
286
285
        self.has_dispatched = False
287
286
        self.request = None
288
287
        self._body_decoder = None
 
288
        self._write_func = write_func
289
289
 
290
290
    def accept_bytes(self, bytes):
291
291
        """Take bytes, and advance the internal state machine appropriately.
351
351
 
352
352
    def _send_response(self, args, body=None):
353
353
        """Send a smart server response down the output stream."""
354
 
        self._out_stream.write(_encode_tuple(args))
355
 
        if body is None:
356
 
            self._out_stream.flush()
357
 
        else:
358
 
            self._send_bulk_data(body, self._out_stream)
 
354
        self._write_func(_encode_tuple(args))
 
355
        if body is not None:
 
356
            assert isinstance(body, str), 'body must be a str'
 
357
            bytes = self._encode_bulk_data(body)
 
358
            self._write_func(bytes)
 
359
            #self._send_bulk_data(body, self._out_stream)
359
360
 
360
361
    def sync_with_request(self, request):
361
362
        self._finished_reading = request.finished_reading
473
474
    which will typically be a LocalTransport looking at the server's filesystem.
474
475
    """
475
476
 
476
 
    def __init__(self, in_file, out_file, backing_transport):
 
477
    def __init__(self, backing_transport):
477
478
        """Construct new server.
478
479
 
479
 
        :param in_file: Python file from which requests can be read.
480
 
        :param out_file: Python file to write responses.
481
480
        :param backing_transport: Transport for the directory served.
482
481
        """
483
 
        self._in = in_file
484
 
        self._out = out_file
485
482
        # backing_transport could be passed to serve instead of __init__
486
483
        self.backing_transport = backing_transport
487
484
        self.finished = False
488
485
 
489
 
    def _send_tuple(self, args):
490
 
        """Send response header"""
491
 
        # ** serialise and write bytes
492
 
        return self._write_and_flush(_encode_tuple(args))
493
 
 
494
486
    def serve(self):
495
487
        """Serve requests until the client disconnects."""
496
488
        # Keep a reference to stderr because the sys module's globals get set to
498
490
        from sys import stderr
499
491
        try:
500
492
            while not self.finished:
501
 
                protocol = SmartServerRequestProtocolOne(self._out,
502
 
                                                         self.backing_transport)
 
493
                protocol = SmartServerRequestProtocolOne(self.backing_transport,
 
494
                                                         self._write_out)
503
495
                self._serve_one_request(protocol)
504
496
        except Exception, e:
505
497
            stderr.write("%s terminating on exception %s\n" % (self, e))
519
511
 
520
512
    def terminate_due_to_error(self):
521
513
        """Called when an unhandled exception from the protocol occurs."""
522
 
        # TODO: This should log to a server log file, but no such thing
523
 
        # exists yet.  Andrew Bennetts 2006-09-29.
524
 
        self._out.close()
525
 
        self.finished = True
 
514
        raise NotImplementedError(self.terminate_due_to_error)
526
515
 
527
516
 
528
517
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
529
518
 
530
 
    def __init__(self, in_socket, out_file, backing_transport):
 
519
    def __init__(self, sock, backing_transport):
531
520
        """Constructor.
532
521
 
533
 
        :param in_socket: the socket the server will read from.  It will be put
 
522
        :param sock: the socket the server will read from.  It will be put
534
523
            into blocking mode.
535
524
        """
536
 
        in_socket.setblocking(True)
537
 
        SmartServerStreamMedium.__init__(
538
 
            self, in_socket, out_file, backing_transport)
 
525
        SmartServerStreamMedium.__init__(self, backing_transport)
539
526
        self.push_back = ''
 
527
        sock.setblocking(True)
 
528
        self.socket = sock
540
529
 
541
530
    def _serve_one_request_unguarded(self, protocol):
542
531
        while protocol.next_read_size():
544
533
                protocol.accept_bytes(self.push_back)
545
534
                self.push_back = ''
546
535
            else:
547
 
                bytes = self._in.recv(4096)
 
536
                bytes = self.socket.recv(4096)
548
537
                if bytes == '':
549
538
                    self.finished = True
550
539
                    return
552
541
        
553
542
        self.push_back = protocol.excess_buffer
554
543
    
 
544
    def terminate_due_to_error(self):
 
545
        """Called when an unhandled exception from the protocol occurs."""
 
546
        # TODO: This should log to a server log file, but no such thing
 
547
        # exists yet.  Andrew Bennetts 2006-09-29.
 
548
        self.socket.close()
 
549
        self.finished = True
 
550
 
 
551
    def _write_out(self, bytes):
 
552
        self.socket.sendall(bytes)
 
553
 
555
554
 
556
555
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
557
556
 
562
561
        :param out_file: Python file to write responses.
563
562
        :param backing_transport: Transport for the directory served.
564
563
        """
565
 
        SmartServerStreamMedium.__init__(self, in_file, out_file, backing_transport)
 
564
        SmartServerStreamMedium.__init__(self, backing_transport)
566
565
        self._in = in_file
567
566
        self._out = out_file
568
567
 
571
570
            bytes_to_read = protocol.next_read_size()
572
571
            if bytes_to_read == 0:
573
572
                # Finished serving this request.
 
573
                self._out.flush()
574
574
                return
575
575
            bytes = self._in.read(bytes_to_read)
576
576
            if bytes == '':
577
577
                # Connection has been closed.
578
578
                self.finished = True
 
579
                self._out.flush()
579
580
                return
580
581
            protocol.accept_bytes(bytes)
581
582
 
 
583
    def terminate_due_to_error(self):
 
584
        # TODO: This should log to a server log file, but no such thing
 
585
        # exists yet.  Andrew Bennetts 2006-09-29.
 
586
        self._out.close()
 
587
        self.finished = True
 
588
 
 
589
    def _write_out(self, bytes):
 
590
        self._out.write(bytes)
 
591
 
582
592
 
583
593
class SmartServerResponse(object):
584
594
    """Response generated by SmartServerRequestHandler."""
865
875
        # propogates to the newly accepted socket.
866
876
        conn.setblocking(True)
867
877
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
868
 
        from_client = conn.makefile('r')
869
 
        to_client = conn.makefile('w')
870
 
        handler = SmartServerSocketStreamMedium(conn, to_client,
871
 
                self.backing_transport)
 
878
        handler = SmartServerSocketStreamMedium(conn, self.backing_transport)
872
879
        connection_thread = threading.Thread(None, handler.serve, name='smart-server-child')
873
880
        connection_thread.setDaemon(True)
874
881
        connection_thread.start()