[Clean code] 클래스

캡슐화

변수와 유틸리티 함수는 가능한 공개하지 않는 편이 낫지만 반드시 숨겨야 한다는 법칙도 없다.
하지만 캡슐화를 풀어주는 결정은 언제나 최후의 수단이다

클래스의 크기 → 항상 작아야 한다

작아야 한다의 기준은 = 클래스가 맡은 책임을 센다.
클래스 이름 = 해당 클래스의 책임을 기술한다

class ClientBluetooth(threading.Thread)

클래스의 목적: 디바이스 이름을 입력 받아 통신을 수립하는 클래스

책임이 두 개다

1. 디바이스의 이름을 입력 받는 책임
2. 통신을 수립하는 책임

class DevInfo

class SearchDev

class MatchServiceToPortNum

class BindToSocket

class CommunicationToDev

DevInfo: 디바이스 정보를 입력 받는 클래스

SearchDev: 주변의 디바이스를 검색하는 클래스

MatchServiceToPortNum:서비스의 포트 번호를 검색하는 클래스

BindToSocket: Device의 address, port를 소캣으로 바인딩하는 클래스

CommunicationToDev: Device와 통신하는 클래스

효과

1. 클래스의 책임을 하나가 되면서 클래스를 변경할 이유가 하나뿐이게 됨으로써 단일 책임의 원칙을 지킬 수 있다
2. 클래스 수정에 폐쇄적이고 확장에는 개방적인 OCP원칙을 지원한다.
   1. Ex) 검색 방법을 바꾸고 싶다면 검색 클래스에 새로운 함수만 추가하면 된다.

응집도

응집도가 높다 = 클래스에 속한 메서드와 변수가 서로 의존하며 논리적인 단위로 묶인다는 의미다

일반적으로 메서드가 클래스 인스턴트 변수를 많이 사용하면 응집도가 높다고 볼 수 있다

public class Stack {
	private int topOfStack = 0;
	List<Integer> elements = new LinkedList<Integer>();

	public int size() { 
		return topOfStack;
	}

	public void push(int element) { 
		topOfStack++; 
		elements.add(element);
	}
	
	public int pop() throws PoppedWhenEmpty { 
		if (topOfStack == 0)
			throw new PoppedWhenEmpty();
		int element = elements.get(--topOfStack); 
		elements.remove(topOfStack);
		return element;
	}
}

응집도가 높다.
클래스 인스턴스 변수를 모든 함수가 사용하고 있기 때문이다.

클래스가 응집력을 잃으면 분리해야 한다.

class ClientBluetooth(threading.Thread):
    #디바이스 네임은 알아야함
    def __init__(self, dev_name, mac_addr=None, call_back=None):
        threading.Thread.__init__(self)
        self._dev_name = dev_name
        self._mac_addr = mac_addr
        self._callback_Func = call_back
        self._port = 30             #변경 필요
        self._thread_flag = False

    def call_service(self, service_name=None, callback=None):
        """
            service_name에 해당하는 포트에 접속해 데이터 수신하고 콜백을 등록한다
            service_name == None -> 기본 포트로 접속한다
            callback == None -> 쓰레드 종료를 flag를 사용해야한다.
        """
        try:
            self.get_target_dev()
            self.get_service(service_name)
            self.get_socket()
            self.set_connection()
        except:
            print("콜 서비스 오류")
        else:
            self.callback_Func = callback
            self.start()

    def get_target_dev(self):
        """
            self._dev_name과 동일한 mac_address를 찾는다

            success -> mac_addr을 설정
            fail -> 통신 연결 종료
        """
        nearby_devices = asyncio.run(self._search_near_dev())
        try:
            dev, = [x for x in nearby_devices if x.name == self._dev_name]
            # dev.address, = list(
            #     filter(lambda x: x.name == self._dev_name, nearby_devices))
            self._mac_addr = dev.address
            print("디바이스 특정 성공", self._mac_addr, self._dev_name)
        except:
            print("디바이스 특정 실패 ")
            raise

    async def _search_near_dev(self):
        nearby_devices = await BleakScanner.discover()
        print("주변의 디바이스 내역", str(
            list(map(lambda x: x.name, nearby_devices))))
        return nearby_devices

    def get_service(self, service_name=None):
        """
            self._dev_name same의 mac_address를 찾는다

            success -> mac_addr을 설정
            fail -> 통신 연결 종료
        """
        services = self._search_service_list()

        if service_name is not None:
            try:
                service, = [
                    service for service in services if service['name'] == service_name]
                self._port = service['port']
            except:
                print("포트 할당 실패")
                raise
        else:
            self._port = 30

    def _search_service_list(self):
        """
            self._dev_name에서 서비스하는 리스트를 읽는다
            return [{service_list}] -> 딕셔너리 배열 
        """
        services = bluetooth.find_service(address=self._mac_addr)
        print('서비스 리스트 ', services)
        return services

    def get_socket(self):
        """
            소캣을 획득하는 함수
        """
        self._socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)

    def set_connection(self):
        """
            선택된 포트로 연결을 수립하는 함수다

            success -> 통신채널 연결
            fail -> 통신 연결 종료
        """
        try:
            self._socket.connect((self._mac_addr, self._port))
        except:
            print("connection Fail")
            raise

    def run(self):
        """
            연결된 통신 채널을 통해서 데이터를 수신, 송신할 수 있다.

            self.thread_flag를 통해 쓰레드를 제어할 수 있다.
            OR
            데이터의 특정한 조건을 통하여 콜백 함수를 실행한다.
        """
        self._thread_flag = True

        while self._thread_flag:
            print(self._socket, self._port)
            data = self._read()
            if data == "특정한 무언가":                                    
                self._callback_Func(data)
                break

            print("server :", data)

        self._thread_flag = False
        self._socket.close()

    def _read(self):
        """
            연결된 소캣에서 데이터를 읽어 온다 
        """
        data = self._socket.recv(1024)
        print("recv data", data)
        return data

    def _write(self, data):
        """
            연결된 소캣에서 데이터를 기록한다 + 
        """
        self._socket.send(data)

    @property
    def thread_flag(self):
        return self._thread_flag
    
    @thread_flag.setter
    def thread_flag(self, flag):
        self.thread_flag = flag

    @property
    def callback_Func(self):
        return self._callback_Func

    @callback_Func.setter
    def callback_Func(self, func):
        self._callback_Func=func

#!/usr/bin/env python3
import bluetooth
import time
import threading
import queue
import asyncio
from bleak import BleakScanner


class DevInfo:
    def __init__(self, dev_name, mac_addr=None, call_back=None, port=30):
        threading.Thread.__init__(self)
        self._dev_name = dev_name
        self._mac_addr = mac_addr
        self._port = port  # 변경 필요

    @property
    def dev_name(self):
        return self._dev_name

    @property
    def mac_addr(self):
        return self._mac_addr

    @property
    def port(self):
        return self._port

    @mac_addr.setter
    def mac_addr(self, mac_addr):
        self._mac_addr = mac_addr

    @mac_addr.setter
    def port(self, port):
        self._port = port


class BluetoothCommucation:
    def __init__(self) -> None:
        pass


class SearchDev(BluetoothCommucation):
    def __init__(self, dev_name) -> None:
        super().__init__()
        self._dev_name = dev_name

    def get_target_dev(self):
        """
            self._dev_name과 동일한 mac_address를 찾는다

            success -> mac_addr을 설정
            fail -> 통신 연결 종료
        """
        nearby_devices = asyncio.run(self._search_near_dev())
        try:
            dev, = [x for x in nearby_devices if x.name == self._dev_name]
            print("Finish get Dev mac addr", dev.address)
            return dev.address
        except:
            print("디바이스 특정 실패 ")
            raise

    async def _search_near_dev(self):
        nearby_devices = await BleakScanner.discover()
        print("주변의 디바이스 내역", str(
            list(map(lambda x: x.name, nearby_devices))))
        return nearby_devices


class MatchServiceToPortNum(BluetoothCommucation):
    def __init__(self, mac_addr, service_name="Test") -> None:
        super().__init__()
        self._service_name = service_name
        self._mac_addr = mac_addr

    def get_service(self):
        """
            self._dev_name same의 mac_address를 찾는다

            success -> mac_addr을 설정
            fail -> 통신 연결 종료
        """
        services = self._search_service_list()

        if self._service_name is not None:
            try:
                service, = [
                    service for service in services if service['name'] == self._service_name]
                return service['port']
            except:
                print("포트 할당 실패")
                raise
        else:
            print("서비스 네임 특정 안됨")
            raise

    def _search_service_list(self):
        """
            self._dev_name에서 서비스하는 리스트를 읽는다
            return [{service_list}] -> 딕셔너리 배열 
        """
        services = bluetooth.find_service(address=self._mac_addr)
        print('서비스 리스트 ', services)
        return services


class BindToSocket(BluetoothCommucation):
    def __init__(self, dev_info) -> None:
        super().__init__()
        self._socket = None
        self._dev_info = dev_info

    def bind_socket(self):
        self._get_socket()
        self._set_connection()
        return self._socket

    def _get_socket(self):
        """
            소캣을 획득하는 함수
        """
        self._socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)

    def _set_connection(self):
        """
            선택된 포트로 연결을 수립하는 함수다

            success -> 통신채널 연결
            fail -> 통신 연결 종료
        """
        try:
            self._socket.connect(
                (self._dev_info.mac_addr, self._dev_info.port))
        except:
            print("connection Fail")
            raise


class CommunicationToDev(BluetoothCommucation, threading.Thread):
    def __init__(self, socket, func=None) -> None:
        super().__init__()
        self._socket = socket
        self.callback_Func = func
        self._thread_flag = False

    def run(self):
        """
            연결된 통신 채널을 통해서 데이터를 수신, 송신할 수 있다.

            self.thread_flag를 통해 쓰레드를 제어할 수 있다.
            OR
            데이터의 특정한 조건을 통하여 콜백 함수를 실행한다.
        """
        self._thread_flag = True

        while self._thread_flag:
            data = self._read()
            if data == "특정한 무언가":
                self._callback_Func(data)
                break

        self._thread_flag = False
        self._socket.close()

    def _read(self):
        """
            연결된 소캣에서 데이터를 읽어 온다 
        """
        data = self._socket.recv(1024)
        print("recv data", data)
        return data

    def _write(self, data):
        """
            연결된 소캣에서 데이터를 기록한다 + 
        """
        self._socket.send(data)

    @property
    def thread_flag(self):
        return self._thread_flag

    @thread_flag.setter
    def thread_flag(self, flag):
        self.thread_flag = flag

    @property
    def callback_Func(self):
        return self._callback_Func

    @callback_Func.setter
    def callback_Func(self, func):
        self._callback_Func = func

기존의 응집도가 낮은 코드를 클래스를 분해하여 응집도를 높이고, 결합도를 낮춤으로써 변경으로부터 격리

Ex)ClientBluetooth.search_dev ()를 바꾸면 클래스의 다른 코드를 망가트릴 위함이 존재했지만, search_dev()를 SearchDev Class로 변경함으로서 변경의 영향을 최소화

변경하기 쉬운 클래스

// 해당 코드는 새로운 SQL문을 지원할 때 손대야 하고, 기존 SQL문을 수정할 때도 손대야 하므로 SRP위반

public class Sql {
	public Sql(String table, Column[] columns)
	public String create()
	public String insert(Object[] fields)
	public String selectAll()
	public String findByKey(String keyColumn, String keyValue)
	public String select(Column column, String pattern)
	public String select(Criteria criteria)
	public String preparedInsert()
	private String columnList(Column[] columns)
	private String valuesList(Object[] fields, final Column[] columns) private String selectWithCriteria(String criteria)
	private String placeholderList(Column[] columns)
}

1. 새로운 sql문을 지원하려면 반드시 sql 클래스를 수정
2. 기존 select문을 수정하려면 sql 클래스를 수정해야 한다

→ 클래스를 변경할 이유가 두 가지이므로 sql클래스 SRP를 위반한다.

	abstract public class Sql {
		public Sql(String table, Column[] columns) 
		abstract public String generate();
	}
	public class CreateSql extends Sql {
		public CreateSql(String table, Column[] columns) 
		@Override public String generate()
	}
	
	public class SelectSql extends Sql {
		public SelectSql(String table, Column[] columns) 
		@Override public String generate()
	}
	
	public class InsertSql extends Sql {
		public InsertSql(String table, Column[] columns, Object[] fields) 
		@Override public String generate()
		private String valuesList(Object[] fields, final Column[] columns)
	}
	
	public class SelectWithCriteriaSql extends Sql { 
		public SelectWithCriteriaSql(
		String table, Column[] columns, Criteria criteria) 
		@Override public String generate()
	}
	
	public class SelectWithMatchSql extends Sql { 
		public SelectWithMatchSql(String table, Column[] columns, Column column, String pattern) 
		@Override public String generate()
	}
	
	public class FindByKeySql extends Sql public FindByKeySql(
		String table, Column[] columns, String keyColumn, String keyValue) 
		@Override public String generate()
	}
	
	public class PreparedInsertSql extends Sql {
		public PreparedInsertSql(String table, Column[] columns) 
		@Override public String generate() {
		private String placeholderList(Column[] columns)
	}
	
	public class Where {
		public Where(String criteria) public String generate()
	}
	
	public class ColumnList {
		public ColumnList(Column[] columns) public String generate()
	}

1. 공개된 인터페이스를 sql에서 파생하는 클래스로 만들었다.

이점:

1. SRP를 지원함으로써 다른 기능에 미칠 영향을 최소화 가능
2. OCP원칙을 지킨다
   1. 새 기능을 추가할 때 기존 코드를 변경하지 않고 시스템을 확장할 수 있다.

변경으로부터 격리

구체적인 클래스 = 상세한 구현을 포함

추상 클래스 = 개념만 포함

결합도를 낮추면 유연성과 재사용성이 노ㅠ아져 변경으로부터 잘 격리가 됨