123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129 |
- #!/usr/bin/python3
- #
- import basic_protocols
- import logging
- import enum
- import time
- import socket
- #
- class ClientProtocol(basic_protocols.Protocol):
- def __init__(self, socket, total_bytes, group_id=None, send_buffer_len=None, use_acceleration=None, custom_data=b'', push_start_cb=None, push_done_cb=None): #wait_until=None
- self.socket = socket
- self.total_bytes = total_bytes
- #self.wait_until = wait_until
- self.send_buffer_len = send_buffer_len
- self.use_acceleration = use_acceleration
- self.custom_data = custom_data
- self.push_start_cb = push_start_cb
- self.push_done_cb = push_done_cb
- self.group_id = group_id if group_id is not None else 0
- # a group id of 0 means no group
- #
- self.states = enum.Enum('CLIENT_CONN_STATES', 'READY_TO_BEGIN SEND_GROUP_ID SEND_CUSTOM_DATA PUSH_DATA DONE') #WAIT
- self.state = self.states.READY_TO_BEGIN
- #
- self.sub_protocol = None
- #
- def _run_iteration(self):
- if self.state is self.states.READY_TO_BEGIN:
- group_id_bytes = self.group_id.to_bytes(8, byteorder='big', signed=False)
- self.sub_protocol = basic_protocols.SendDataProtocol(self.socket, group_id_bytes)
- self.state = self.states.SEND_GROUP_ID
- #
- if self.state is self.states.SEND_GROUP_ID:
- if self.sub_protocol.run():
- self.sub_protocol = basic_protocols.SendDataProtocol(self.socket, self.custom_data)
- self.state = self.states.SEND_CUSTOM_DATA
- #
- #
- if self.state is self.states.SEND_CUSTOM_DATA:
- if self.sub_protocol.run():
- #self.state = self.states.WAIT
- self.sub_protocol = basic_protocols.PushDataProtocol(self.socket, self.total_bytes,
- send_buffer_len=self.send_buffer_len,
- use_acceleration=self.use_acceleration,
- push_start_cb=self.push_start_cb,
- push_done_cb=self.push_done_cb)
- self.state = self.states.PUSH_DATA
- #
- #
- '''
- if self.state is self.states.WAIT:
- if self.wait_until is not None:
- time.sleep(self.wait_until-time.time())
- #
- if self.wait_until is None or time.time() >= self.wait_until:
- self.sub_protocol = basic_protocols.PushDataProtocol(self.socket, self.total_bytes,
- send_buffer_len=self.send_buffer_len,
- use_acceleration=self.use_acceleration,
- push_start_cb=self.push_start_cb,
- push_done_cb=self.push_done_cb)
- self.state = self.states.PUSH_DATA
- #
- #
- '''
- if self.state is self.states.PUSH_DATA:
- if self.sub_protocol.run():
- self.state = self.states.DONE
- #
- #
- if self.state is self.states.DONE:
- return True
- #
- return False
- #
- #
- class ServerProtocol(basic_protocols.Protocol):
- def __init__(self, socket, conn_id, group_id_callback=None, bandwidth_callback=None, use_acceleration=None):
- self.socket = socket
- self.conn_id = conn_id
- self.group_id_callback = group_id_callback
- self.bandwidth_callback = bandwidth_callback
- self.use_acceleration = use_acceleration
- #
- self.states = enum.Enum('SERVER_CONN_STATES', 'READY_TO_BEGIN RECV_GROUP_ID RECV_CUSTOM_DATA PULL_DATA DONE')
- self.state = self.states.READY_TO_BEGIN
- #
- self.sub_protocol = None
- self.custom_data = None
- #
- def _run_iteration(self):
- if self.state is self.states.READY_TO_BEGIN:
- self.sub_protocol = basic_protocols.ReceiveDataProtocol(self.socket)
- self.state = self.states.RECV_GROUP_ID
- #
- if self.state is self.states.RECV_GROUP_ID:
- if self.sub_protocol.run():
- group_id = int.from_bytes(self.sub_protocol.received_data, byteorder='big', signed=False)
- if group_id == 0:
- # a group of 0 means no group
- group_id = None
- #
- self.group_id_callback(self.conn_id, group_id)
- self.sub_protocol = basic_protocols.ReceiveDataProtocol(self.socket)
- self.state = self.states.RECV_CUSTOM_DATA
- #
- #
- if self.state is self.states.RECV_CUSTOM_DATA:
- if self.sub_protocol.run():
- self.custom_data = self.sub_protocol.received_data
- #
- self.sub_protocol = basic_protocols.PullDataProtocol(self.socket, use_acceleration=self.use_acceleration)
- self.state = self.states.PULL_DATA
- #
- #
- if self.state is self.states.PULL_DATA:
- if self.sub_protocol.run():
- if self.bandwidth_callback:
- #self.bandwidth_callback(self.conn_id, self.sub_protocol.data_size, self.sub_protocol.time_of_first_byte, self.sub_protocol.time_of_last_byte, self.sub_protocol.calc_transfer_rate(), self.sub_protocol.byte_counter, self.sub_protocol.byte_counter_start_time)
- self.bandwidth_callback(self.conn_id, self.custom_data, self.sub_protocol.data_size, self.sub_protocol.time_of_first_byte, self.sub_protocol.time_of_last_byte, self.sub_protocol.calc_transfer_rate(), self.sub_protocol.deltas)
- #
- self.state = self.states.DONE
- #
- #
- if self.state is self.states.DONE:
- return True
- #
- return False
- #
- #
|