소켓 프로그래밍에서 콜백을 이용하여 데이터 가공하기
안녕하세요. 저는 현재 데이터 관리 업무를 하고 있는 주니어 개발자입니다.
최근 절차적으로 구현 된 소켓 프로그래밍 구조를 리팩터링하는 과정에서 겪은 내용을 공유 하고자 글을 남깁니다.
기존 소스코드는 모듈화 된 함수 한 개에 비즈니스 로직이 몰려 있는 형태로 코드를 수정 할 때 마다 회귀버그를 종종 마주한 적 있었습니다. 그래서 수정이 용이하고 테스트 하기 쉬운 코드를 작성하기 위해 많은 고민을 해보았는데요.
그 고민 속 제가 접근한 방식은 소켓 서버 객체에 데이터 핸들러 역할을 하는 콜백 함수를 인자로 넘겨, 실제 비즈니스 로직이 수정 되어도 콜백 내부에서만 수정 되어 호출 하는 소켓 객체에서는 아무런 영향을 받지 않을 수 있도록 의도 하여 작성하였습니다.
소켓 객체 생성
소켓 서버의 내부 프로세스 선언
class SocketServer:
callback_parse = None
callback_convert = None
def __init__(self, info: dict):
# note: 소켓 서버를 실행 할 설정 값이 담겨 있는 데이터 클래스 호출
self.realtime_receive_info = Config(info)
@staticmethod
def server_listen(_socket, host, port):
_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
_socket.bind((host, port))
_socket.listen()
logger.info(f"receiver process message listening on: {host} / {port}")
def message_receive_process(self, client_socket) -> Tuple[Callable, str]:
"""
소켓 서버에서 클라이언트 메세지 수신 및 응답 프로세스
"""
pass
def convert_process(self, realtime_receive_data: bytes):
"""
정상적인 메세지 수신 일 때 데이터 가공
"""
if self.callback_parse:
parse_message_dict = self.callback_parse(realtime_receive_data)
logger.info(parse_message_dict)
if self.callback_convert:
sales_list_data = self.callback_convert(parse_message_dict)
return parse_message_dict
@staticmethod
def invalid_data_receive_handler(*args):
"""
비정상적인 메세지 수신으로 프로세스를 시작 지점으로 이동
"""
raise ValueError("Exception by invalid_data_receive_handler: received invalid data!!!")
def run_server(self):
"""
실시간 수신 서버 실행
"""
with socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM) as sock:
self.server_listen(sock, self.realtime_receive_info.host, self.realtime_receive_info.port)
while 1:
client_socket, (client_host, client_port) = sock.accept()
logger.info(f"Client(IP : {client_host}, Port : {client_port})")
try:
# note: 데이터 가공을 위한 콜백 메서드, 실시간 수신 데이터 언패킹
next_process_method, realtime_receive_data = self.message_receive_process(client_socket)
next_process_method(realtime_receive_data)
except TimeOutException as e:
logger.error(f"Timeout occurred! >> error message: {e}")
signal.alarm(0)
continue
except Exception as e:
logger.error(e, exc_info=True)
signal.alarm(0)
continue
signal.alarm(0)
소켓 서버에서 데이터 핸들링을 처리하는 객체
class RealtimeMessageReceiver(SocketServer):
def __init__(self, info, callback_parse=None, callback_convert=None):
super().__init__(info)
self.callback_parse = callback_parse
self.callback_convert = callback_convert
def check_receive_data_length(self, data: bytes) -> bool:
"""
데이터 정상 수신 여부: 데이터 길이 검증
"""
if data == "" or len(data) != self.realtime_receive_info.receive_info.data_length:
return False
else:
return True
def response_message(self, message: bytes, success_flag: bool) -> bytes:
"""
클라이언트 응답 코드
:param message: 실시간 수신 데이터
:param success_flag: 수신 상태 - 정상 / 실패
:return: 응답 코드가 포함 된 바이트 타입의 데이터
"""
# response_code 생성 로직
return response_code
@TimeOutHandler(10)
def message_receive_process(self, client_socket):
get_response_code = None
message = client_socket.recv(1024)
try:
if self.check_receive_data_length(message):
get_response_code = self.response_message(message, True)
next_step = self.convert_process
next_step_argument = message
else:
get_response_code = self.response_message(message, False)
next_step = self.invalid_data_receive_handler
next_step_argument = ""
return next_step, next_step_argument
finally:
client_socket.sendall(get_response_code)
signal.alarm(0)
logger.info("Close client socket")
client_socket.close()
실시간 통신을 할 수 있는 TCP 소켓 서버 입니다. 간단하게 구조를 설명 드리자면 이렇습니다.
- 소켓 객체를 생성 하고, 명시된 IP 주소를 바인딩 합니다.
- 소켓 서버에서 클라이언트가 접속 할 수 있도록 요청을 받을 준비를 합니다.
- 무한 루프를 통해 클라이언트의 접속 요청이 있는지 지속적으로 확인합니다.
- 클라이언트의 메세지를 수신 합니다.
- 클라이언트는 접속을 종료 시킵니다.
- 클라이언트의 메세지가 정상인지 검증합니다. 6-1. 정상적인 데이터라면 데이터를 필요한 형태로 가공합니다. 6-2. 비정상적인 데이터라면 에러 메세지와 함께 프로그램은 [3]번째 상태로 돌아가게 됩니다.
위 구조에서 기존 코드라면 하나의 함수가 파서, 또 다른 하나의 함수가 컨버터 이렇게 2가지 함수를 통해 비즈니스 로직이 몰려 있었던 터라, 그 코드를 수정 하면 호출 하는 부분에서도 수정이 필요하는 회귀 케이스가 되었습니다.
그렇다면 외부에서 해당 함수를 주입하고, 함수는 내부에서 호출만 할 수 있도록 구조를 만들어보겠습니다.
콜백 객체 만들기
파서 프로세스 콜백 객체 만들기
class RealtimeParserCallback:
def __init__(self, args: Namespace, client: Database, parse_schema):
self.args = args
self.client = client
self.parse_schema = parse_schema
def __call__(self, *args, **kwargs):
return self.callback_method(*args, **kwargs)
@property
def realtime_parser_instance(self) -> RealtimeMessageParser:
"""
실시간 수신 데이터 파서 클래스
"""
return RealtimeMessageParser(self.parse_schema)
@property
def realtime_query_manager(self) -> RealtimeQueryManager:
collection: Dict[str, str] = self.parse_schema["collections"]
return RealtimeQueryManager(self.client, collection)
@property
def callback_method(self):
return self.get_realtime_parse_process_callback_method()
def get_realtime_parse_process_callback_method(self) -> Callable:
"""
데이터 가공을 위한 인스턴스화 클로저
"""
def parse_process_callback_method(data: bytes) -> Dict[str, Union[str, int]]:
"""
데이터 가공 프로세스 실행
"""
# note: 데이터 파싱
raw_data_result = self.realtime_parser_instance.create_raw_data_format(data)
parse_result = self.realtime_parser_instance.parse_data(data)
# note: raw_format 데이터 저장
raw_data_insert_result = self.realtime_query_manager.insert_one_realtime_raw(raw_data_result)
# note: 원본 데이터 저장 한 mongodb Object ID 값을 가공 데이터에 추가
parse_result_dict = add_row_data_object_id(parse_result, raw_data_insert_result.inserted_id)
saved_data = self.realtime_query_manager.insert_one_realtime(parse_result_dict) # note: 가공 데이터 저장
return saved_data
return parse_process_callback_method바이트 형태의 데이터를 파싱하여 문자열 형태로 데이터베이스에 저장 하는 과정이 담긴 콜백 함수인데요. 지금은 클로저 형태를 띄고 있지만, 다시 보니 굳이 클로저를 넘기지 않고 메서드의 참조값만 넘겨도 될 것 같다고 생각이 드네요.
이렇게 프로젝트를 하면서 작업한 내용을 조금씩 공유를 할 때 생각이 정리 되는 것 같아 글을 쓰지 않을 이유가 없다는 것을 대변할 수 있는 것 같습니다.
글을 남기면서 즉석으로 수정을 해 보자면 제 말의 뜻은 이렇습니다.
class RealtimeParserCallback:
def __init__(self, args: Namespace, client: Database, parse_schema):
self.args = args
self.client = client
self.parse_schema = parse_schema
def __call__(self, *args, **kwargs):
return self.parse_process_callback_method(*args, **kwargs)
@property
def realtime_parser_instance(self) -> RealtimeMessageParser:
"""
실시간 수신 데이터 파서 클래스
"""
return RealtimeMessageParser(self.parse_schema)
@property
def realtime_query_manager(self) -> RealtimeQueryManager:
collection: Dict[str, str] = self.parse_schema["collections"]
return RealtimeQueryManager(self.client, collection)
def parse_process_callback_method(data: bytes) -> Dict[str, Union[str, int]]:
"""
데이터 가공 프로세스 실행
"""
# note: 데이터 파싱
raw_data_result = self.realtime_parser_instance.create_raw_data_format(data)
parse_result = self.realtime_parser_instance.parse_data(data)
# note: raw_format 데이터 저장
raw_data_insert_result = self.realtime_query_manager.insert_one_realtime_raw(raw_data_result)
# note: 원본 데이터 저장 한 mongodb Object ID 값을 가공 데이터에 추가
parse_result_dict = add_row_data_object_id(parse_result, raw_data_insert_result.inserted_id)
saved_data = self.realtime_query_manager.insert_one_realtime(parse_result_dict) # note: 가공 데이터 저장
return saved_data
# 객체 호출
callback_parser = RealtimeParserCallback(**schemas)
realtime_socket_instance(callback_parser)
# 실시간 서버 내에서 데이터 가공
result = callback_parser(data)
네, 뭐 이런 느낌이 되지 않을까 싶습니다. 이렇든 저렇든 소켓 객체에 이러한 데이터 핸들링을 하는 객체를 콜러블한 구조를 만들어 넘겨준다면 호출 하는 입장에서는 늘 callback_parser(data) 만 호출하면 되겠죠. 그 콜백 내부에서만 로직을 수정하면 수정하거나, 테스트를 하는데도 문제가 없게 됩니다.
네, 이런 콜백 함수로 만들어진 파서 객체 외에도 데이터 컨버팅을 처리하는 컨버터 파서도 존재 하는데요. 구조는 이와 같습니다.
호출
def main():
realtime_receiver_args = "소켓 서버가 실행 될 때 필요한 정보가 담긴 파일"
db_client = get_mongo_database("서버 환경", db_name)
from src.receiver import main_receiver # note: 전역에 할당 된 logger를 초기화 이후 선언 하기 위해 동적 임포트
callback_parse = RealtimeParserCallback("argsparse", db_client, schema.realtime_parser_schema)
callback_convert = RealtimeConverterCallback("argsparse", db_client, schema.converter_schema)
main_receiver.run_receiver(realtime_receiver_args, callback_parse, callback_convert, schema.receiver_schema)소켓서버를 실행 할 때 위 처럼 콜백을 넘겨주는 형태로 구조를 갖추는 방법을 처음부터 의도 해 왔는데 의도한 바 대로 구조가 만들어진 것 같습니다.
다음엔 이렇게 콜백을 통해 넘겨준 소켓 객체를 어떻게 테스트 할 것인가에 대해 남겨보는 시간을 가져봐야 할 것 같습니다. 테스트 코드 작성이 최근에 끝났고 그 내용에 대한 정답이 없다보니 굉장히 많은 시간을 쏟고 있는데 아마 여럿 단계를 거쳐 수정과 수정을 반복 하게 될 것 같습니다.
끝맺음
결론은 이 객체 내부에서 동작하는 함수 구조가 굉장히 단순하게 묶여있다면 과연 수정했을 때 회귀케이스가 되지 않을까? 이런 함수는 어떻게 테스트를 해야 할까? 만약, 내부에서 여럿 모듈과 상호작용 하고 있다면 어떻게 해야할까, 코드가 실행하는 중간 에러가 나는 경우 어떻게 디버깅을 할 것인가 등 다양한 고민을 통해 객체 형태로 변경 시키고 그 객체의 내부 메서드를 호출 할 수 있는 콜백 구조를 만들어 호출하는 입장에서 영향을 받지 않도록 고민 하던 결과를 공유 해보았습니다.
감사합니다.