DEVOCEAN

<dependency>
  <groupId>com.alipay.sofa</groupId>
  <artifactId>jraft-core</artifactId>
  <version>1.3.15.bugfix</version>
</dependency>

public class PriorityQueueStateMachine extends StateMachineAdapter {
    // 우선순위 큐 구현을 위한 힙 자료구조
    private final PriorityQueue<Element> queue = new PriorityQueue<>((e1, e2) ->
            Integer.compare(e2.getPriority(), e1.getPriority()));

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    /**
     * Raft 로그를 적용하여 상태 머신을 업데이트합니다.
     *
     * @param iter 로그 항목에 대한 반복자
     */
    @Override
    @SuppressWarnings("unchecked")
    public void onApply(Iterator iter) {
        lock.writeLock().lock(); // 쓰기 락 획득
        try {
            while (iter.hasNext()) { // 로그 항목을 순회
                PriorityQueueRequest op = deserialize(iter.getData().array()); // 로그 데이터에서 PriorityQueueRequest 역직렬화
                try {
                    switch (op.getType()) { // 요청 타입에 따라 처리
                        case ENQUEUE: // 삽입 연산
                            queue.add(op.getElement()); // 큐에 요소 삽입
                            PriorityQueueClosure<Boolean> enqueueClosure = null; // 클로저 초기화
                            if (iter.done() != null) { // 클로저가 존재하는 경우
                                enqueueClosure = (PriorityQueueClosure<Boolean>) iter.done(); // 클로저 캐스팅
                            }
                            if (enqueueClosure != null) { // 클로저가 존재하는 경우
                                enqueueClosure.setResponse(true); // 응답 설정
                                enqueueClosure.run(Status.OK()); // 클로저 실행
                            }
                            LOG.info("Enqueued element: {}", op.getElement());
                            break;
                        case DEQUEUE: // 삭제 연산
                            PriorityQueueClosure<Element> dequeueClosure = null; // 클로저 초기화
                            if (iter.done() != null) { // 클로저가 존재하는 경우
                                dequeueClosure = (PriorityQueueClosure<Element>) iter.done(); // 클로저 캐스팅
                            }
                            if (queue.isEmpty()) { // 큐가 비어있는 경우
                                if (dequeueClosure != null) { // 클로저가 존재하는 경우
                                    dequeueClosure.setResponse(null); // 응답 설정 (null)
                                    dequeueClosure.run(Status.OK()); // 클로저 실행
                                }
                                LOG.info("Dequeue operation on empty queue");
                            } else { // 큐가 비어있지 않은 경우
                                Element element = queue.poll(); // 큐에서 요소 삭제 (가장 우선순위가 높은 요소)
                                if (dequeueClosure != null) { // 클로저가 존재하는 경우
                                    dequeueClosure.setResponse(element); // 응답 설정 (삭제된 요소)
                                    dequeueClosure.run(Status.OK()); // 클로저 실행
                                }
                                LOG.info("Dequeued element: {}", element);
                            }
                            break;
                        case PEEK: // 조회 연산
                            PriorityQueueClosure<Element> peekClosure = null; // 클로저 초기화
                            if (iter.done() != null) { // 클로저가 존재하는 경우
                                peekClosure = (PriorityQueueClosure<Element>) iter.done(); // 클로저 캐스팅
                            }
                            if (queue.isEmpty()) { // 큐가 비어있는 경우
                                if (peekClosure != null) { // 클로저가 존재하는 경우
                                    peekClosure.setResponse(null); // 응답 설정 (null)
                                    peekClosure.run(Status.OK()); // 클로저 실행
                                }
                                LOG.info("Peek operation on empty queue");
                            } else { // 큐가 비어있지 않은 경우
                                Element element = queue.peek(); // 큐에서 요소 조회 (가장 우선순위가 높은 요소)
                                if (peekClosure != null) { // 클로저가 존재하는 경우
                                    peekClosure.setResponse(element); // 응답 설정 (조회된 요소)
                                    peekClosure.run(Status.OK()); // 클로저 실행
                                }
                                LOG.info("Peeked element: {}", element);
                            }
                            break;
                    }
                } catch (Exception e) { // 예외 발생 시
                    LOG.error("Error in processing request", e);
                } finally {
                    iter.next(); // 다음 로그 항목으로 이동
                }
            }
        } finally {
            lock.writeLock().unlock(); // 쓰기 락 해제
        }
    }
    // 스냅샷 생성 및 로드 구현 생략...
}

public class PriorityQueueRequest implements Serializable {
    private final OperationType type;
    private final Element element;

    public PriorityQueueRequest(OperationType type, Element element) {
        this.type = type;
        this.element = element;
    }

    public OperationType getType() {
        return type;
    }

    public Element getElement() {
        return element;
    }
}

public class PriorityQueueServer {
    private final RaftGroupService raftGroupService;
    private Node node;
    private final PriorityQueueStateMachine stateMachine;
    private final String dataPath;
    private final String groupId;
    private final PeerId serverId;
    private final RpcServer rpcServer;
    private final AtomicBoolean started = new AtomicBoolean(false);

    public PriorityQueueServer(String dataPath, String groupId, PeerId serverId, List<PeerId> peerIds) throws IOException {
        this.dataPath = dataPath;
        this.groupId = groupId;
        this.serverId = serverId;
        // RPC 서버 초기화
        this.rpcServer = RaftRpcServerFactory.createRaftRpcServer(serverId.getEndpoint());
        // Raft 옵션 설정
        RaftOptions raftOptions = new RaftOptions();
        // 상태 머신 생성
        this.stateMachine = new PriorityQueueStateMachine();
        // 노드 옵션 설정
        NodeOptions nodeOptions = new NodeOptions();
        nodeOptions.setFsm(this.stateMachine);
        nodeOptions.setLogUri(dataPath + File.separator + "log");
        nodeOptions.setRaftMetaUri(dataPath + File.separator + "raft_meta");
        nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
        nodeOptions.setInitialConf(new Configuration(peerIds));
        nodeOptions.setSnapshotIntervalSecs(30);
        // Raft 그룹 서비스 및 노드 초기화
        this.raftGroupService = new RaftGroupService(groupId, serverId, nodeOptions, rpcServer);
    }
    *// 메서드 생략...*
 }

public class PriorityQueueClient {
    private final CliClientServiceImpl cliClientService;
    private final String groupId;
    private PeerId leaderId;
    private final int rpcTimeoutMs = 5000;

    /**
     * 우선순위 큐에 대한 특정 연산(enqueue, dequeue, peek)을 수행하고 결과를 반환합니다.
     *
     * @param type    수행할 연산 타입 (OperationType enum)
     * @param element 연산에 필요한 요소 (enqueue의 경우 추가할 요소, dequeue/peek의 경우 null)
     * @param <T>     반환 타입 (enqueue는 boolean, dequeue/peek는 Element)
     * @return 연산 결과
     * @throws Exception 연산 실패 시 예외 발생
     */
    @SuppressWarnings("unchecked")
    private <T> T processOperation(OperationType type, Element element) throws Exception {
        if (!refreshLeader()) {
            throw new IllegalStateException("Leader not available"); // 리더가 없으면 예외 발생
        }

        // 요청 객체를 직렬화
        final BytesValue msg = BytesValue.of(ByteString.copyFrom(serialize(new PriorityQueueRequest(type, element))));
        CompletableFuture<T> future = new CompletableFuture<>(); // 비동기 결과를 처리하기 위한 CompletableFuture 생성

        // RPC 호출 결과를 처리하기 위한 ClosureAdapter 생성
        RpcResponseClosureAdapter<BytesValue> closureAdapter = new RpcResponseClosureAdapter<>() {
            @Override
            public void run(Status status) {
                if (status.isOk()) { // RPC 호출 성공 시
                    BytesValue response = getResponse(); // 응답 획득
                    byte[] responseData = response.getValue().toByteArray(); // 응답 데이터 획득
                    PriorityQueueResponse<?> queueResponse = (PriorityQueueResponse<?>) deserialize(responseData); // 응답 데이터 역직렬화
                    future.complete((T) queueResponse.getResult()); // future에 결과를 설정
                } else { // RPC 호출 실패 시
                    future.completeExceptionally(new Throwable(status.getErrorMsg())); // future에 예외를 설정
                }
            }
        };

        // 리더에게 RPC 호출
        cliClientService.invokeWithDone(
                leaderId.getEndpoint(), // 리더의 엔드포인트
                msg,     // Message 타입으로 래핑된 객체
                closureAdapter, // 결과를 처리할 ClosureAdapter
                rpcTimeoutMs // RPC 타임아웃
        );

        // closure에서 결과 대기 및 반환
        return future.get(rpcTimeoutMs, TimeUnit.MILLISECONDS); // future에서 결과를 얻어 반환 (타임아웃 설정)
    }

    /**
     * 우선순위 큐에 요소를 삽입합니다.
     *
     * @param value    요소의 값
     * @param priority 요소의 우선순위
     * @return 삽입 성공 여부
     * @throws Exception 연산 실패 시 예외 발생
     */
    public boolean enqueue(String value, int priority) throws Exception {
        return processOperation(OperationType.ENQUEUE, new Element(value, priority)); // enqueue 연산 수행
    }

    /**
     * 우선순위 큐에서 가장 높은 우선순위를 가진 요소를 삭제하고 반환합니다.
     *
     * @return 삭제된 요소
     * @throws Exception 연산 실패 시 예외 발생
     */
    public Element dequeue() throws Exception {
        return processOperation(OperationType.DEQUEUE, null); // dequeue 연산 수행
    }

    /**
     * 우선순위 큐에서 가장 높은 우선순위를 가진 요소를 반환합니다 (삭제하지 않음).
     *
     * @return 큐의 head 에 있는 요소
     * @throws Exception 연산 실패 시 예외 발생
     */
    public Element peek() throws Exception {
        return processOperation(OperationType.PEEK, null); // peek 연산 수행
    }

    // 메서드 생략...
}