vldpyatkov commented on code in PR #927: URL: https://github.com/apache/ignite-3/pull/927#discussion_r921105769
########## modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java: ########## @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.replicator; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.replicator.message.ReplicaMessageGroup; +import org.apache.ignite.internal.replicator.message.ReplicaRequest; +import org.apache.ignite.internal.replicator.message.ReplicaResponse; +import org.apache.ignite.internal.storage.MvPartitionStorage; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.lang.IgniteException; +import org.apache.ignite.lang.IgniteInternalException; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.lang.NodeStoppingException; +import org.apache.ignite.network.ClusterService; +import org.apache.ignite.network.NetworkAddress; +import org.apache.ignite.network.NetworkMessage; +import org.apache.ignite.network.NetworkMessageHandler; +import org.apache.ignite.raft.client.service.RaftGroupService; +import org.jetbrains.annotations.Nullable; + +/** + * The replica service. + */ +public class ReplicaManager implements IgniteComponent { + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock(); + + /** Prevents double stopping the component. */ + private final AtomicBoolean stopGuard = new AtomicBoolean(); + + /** Cluster network service. */ + private final ClusterService clusterNetSvc; + + /** Partition storage. */ + private final MvPartitionStorage mvDataStorage; + + /** Replicas. */ + private final ConcurrentHashMap<String, Replica> replicas = new ConcurrentHashMap<>(); + + /** + * Constructor for replica service. + * + * @param clusterNetSvc Cluster network service. + * @param mvDataStorage Partition storage. + */ + public ReplicaManager(ClusterService clusterNetSvc, MvPartitionStorage mvDataStorage) { + this.clusterNetSvc = clusterNetSvc; + this.mvDataStorage = mvDataStorage; + + clusterNetSvc.messagingService().addMessageHandler(ReplicaMessageGroup.class, new NetworkMessageHandler() { + @Override + public void onReceived(NetworkMessage message, NetworkAddress senderAddr, @Nullable Long correlationId) { + if (!busyLock.enterBusy()) { + throw new IgniteException(new NodeStoppingException()); + } + + try { + assert message instanceof ReplicaRequest : IgniteStringFormatter.format("Unexpected message [message={}]", message); + + ReplicaRequest request = (ReplicaRequest) message; + + Replica replica = replicas.get(request.partitionId()); + + if (replica == null) { + //TODO:IGNITE-17255 Sent an exceptional response to the client side when the replica is absent. + } + + ReplicaResponse resp = replica.processRequest(request); + + clusterNetSvc.messagingService().respond(senderAddr, resp, correlationId); + } finally { + busyLock.leaveBusy(); + } + } + }); + } + + /** + * Gets a replica by its partition id. + * + * @param partitionId Partition id. + * @return Instance of the replica or {@code null} if the replica is not started. + */ + public Replica replica(String partitionId) { + if (!busyLock.enterBusy()) { + throw new IgniteException(new NodeStoppingException()); + } + + try { + return replicas.get(partitionId); + } finally { + busyLock.leaveBusy(); + } + } + + /** + * Starts replica server. + * + * @param partId Partition id. + * @param raftClient Raft client. + * @return Replica. + */ + public Replica startReplica(String partId, RaftGroupService raftClient) { Review Comment: I will change the Javadoc, but I don't want parametrized this method with a listener, because we have only one implementation. If that method will be required in the future, I think, we will be able to extract a listener interface and add a new method for this case. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
