<commit_before>/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed 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 net.kuujo.copycat.cluster.internal;

import net.kuujo.copycat.EventListener;
import net.kuujo.copycat.cluster.*;
import net.kuujo.copycat.protocol.Protocol;
import net.kuujo.copycat.raft.RaftContext;
import net.kuujo.copycat.raft.protocol.Request;
import net.kuujo.copycat.raft.protocol.Response;
import net.kuujo.copycat.resource.ResourceContext;
import net.kuujo.copycat.util.Managed;
import net.kuujo.copycat.util.concurrent.Futures;
import net.kuujo.copycat.util.internal.Assert;
import net.kuujo.copycat.util.serializer.KryoSerializer;
import net.kuujo.copycat.util.serializer.Serializer;

import java.nio.ByteBuffer;
import java.util.Observable;
import java.util.Observer;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * Default cluster implementation.
 *
 * @author <a href="http://github.com/kuujo"><NAME></a>
 */
public class ManagedCluster implements Cluster, Managed<Cluster>, Observer {
  private final RaftContext raft;
  private final ManagedMembers members;
  private final Set<EventListener<ElectionEvent>> listeners = new CopyOnWriteArraySet<>();
  private final Serializer serializer = new KryoSerializer();
  private String lastLeader;
  private long lastTerm;

  public ManagedCluster(Protocol protocol, ResourceContext context) {
    this.raft = Assert.isNotNull(context, "context").raft();
    this.members = new ManagedMembers(protocol, context);
  }

  @Override
  public void update(Observable o, Object arg) {
    RaftContext raft = (RaftContext) o;
    if (raft.getTerm() > lastTerm && (lastLeader == null || (raft.getLeader() != null && !lastLeader.equals(raft.getLeader())))) {
      lastTerm = raft.getTerm();
      lastLeader = raft.getLeader();
      listeners.forEach(l -> l.accept(new ElectionEvent(ElectionEvent.Type.COMPLETE, lastTerm, member(lastLeader))));
    }
  }

  @Override
  public Member leader() {
    return members.members.get(raft.getLeader());
  }

  @Override
  public long term() {
    return raft.getTerm();
  }

  @Override
  public ManagedLocalMember member() {
    return (ManagedLocalMember) members.members.get(raft.getLocalMember().uri());
  }

  @Override
  public Member member(String uri) {
    return members.members.get(uri);
  }

  @Override
  public Members members() {
    return members;
  }

  @Override
  public <T> Cluster broadcast(String topic, T message) {
    members.forEach(m -> {
      if (m instanceof ManagedRemoteMember) {
        m.send(topic, message);
      }
    });
    return this;
  }

  @Override
  public Cluster addMembershipListener(EventListener<MembershipEvent> listener) {
    members.addListener(listener);
    return this;
  }

  @Override
  public Cluster removeMembershipListener(EventListener<MembershipEvent> listener) {
    members.removeListener(listener);
    return this;
  }

  @Override
  public Cluster addElectionListener(EventListener<ElectionEvent> listener) {
    listeners.add(listener);
    return this;
  }

  @Override
  public Cluster removeElectionListener(EventListener<ElectionEvent> listener) {
    listeners.remove(listener);
    return this;
  }

  /**
   * Registers handlers for the context.
   */
  private void registerHandlers() {
    member().registerInternalHandler(InternalTopics.JOIN, wrapInboundRequest(raft::join));
    member().registerInternalHandler(InternalTopics.PROMOTE, wrapInboundRequest(raft::promote));
    member().registerInternalHandler(InternalTopics.LEAVE, wrapInboundRequest(raft::leave));
    member().registerInternalHandler(InternalTopics.SYNC, wrapInboundRequest(raft::sync));
    member().registerInternalHandler(InternalTopics.POLL, wrapInboundRequest(raft::poll));
    member().registerInternalHandler(InternalTopics.VOTE, wrapInboundRequest(raft::vote));
    member().registerInternalHandler(InternalTopics.APPEND, wrapInboundRequest(raft::append));
    member().registerInternalHandler(InternalTopics.QUERY, wrapInboundRequest(raft::query));
    member().registerInternalHandler(InternalTopics.COMMAND, wrapInboundRequest(raft::command));
    raft.joinHandler(request -> wrapOutboundRequest(InternalTopics.JOIN, request));
    raft.promoteHandler(request -> wrapOutboundRequest(InternalTopics.PROMOTE, request));
    raft.leaveHandler(request -> wrapOutboundRequest(InternalTopics.LEAVE, request));
    raft.syncHandler(request -> wrapOutboundRequest(InternalTopics.SYNC, request));
    raft.pollHandler(request -> wrapOutboundRequest(InternalTopics.POLL, request));
    raft.voteHandler(request -> wrapOutboundRequest(InternalTopics.VOTE, request));
    raft.appendHandler(request -> wrapOutboundRequest(InternalTopics.APPEND, request));
    raft.queryHandler(request -> wrapOutboundRequest(InternalTopics.QUERY, request));
    raft.commandHandler(request -> wrapOutboundRequest(InternalTopics.COMMAND, request));
  }

  /**
   * Unregisters handlers for the context.
   */
  private void unregisterHandlers() {
    member().unregisterInternalHandler(InternalTopics.JOIN);
    member().unregisterInternalHandler(InternalTopics.PROMOTE);
    member().unregisterInternalHandler(InternalTopics.LEAVE);
    member().unregisterInternalHandler(InternalTopics.SYNC);
    member().unregisterInternalHandler(InternalTopics.POLL);
    member().unregisterInternalHandler(InternalTopics.VOTE);
    member().unregisterInternalHandler(InternalTopics.APPEND);
    member().unregisterInternalHandler(InternalTopics.QUERY);
    member().unregisterInternalHandler(InternalTopics.COMMAND);
    raft.joinHandler(null);
    raft.promoteHandler(null);
    raft.leaveHandler(null);
    raft.syncHandler(null);
    raft.pollHandler(null);
    raft.voteHandler(null);
    raft.appendHandler(null);
    raft.queryHandler(null);
    raft.commandHandler(null);
  }

  /**
   * Wraps an inbound context call.
   */
  private <T, U> MessageHandler<ByteBuffer, ByteBuffer> wrapInboundRequest(MessageHandler<T, U> handler) {
    return message -> CompletableFuture.completedFuture(null)
        .thenCompose(v -> handler.apply(serializer.readObject(message)))
        .thenApply(serializer::writeObject);
  }

  /**
   * Wraps an outbound context call.
   */
  private <T extends Request, U extends Response> CompletableFuture<U> wrapOutboundRequest(String topic, T request) {
    ManagedMember<?> member = members.members.get(request.uri());
    if (member != null) {
      return member.sendInternal(topic, serializer.writeObject(request)).thenApply(serializer::readObject);
    }
    return Futures.exceptionalFuture(new ClusterException("Unknown member URI"));
  }

  @Override
  public CompletableFuture<Cluster> open() {
    raft.addObserver(this);
    registerHandlers();
    return members.open().thenApply(v -> this);
  }

  @Override
  public boolean isOpen() {
    return members.isOpen();
  }

  @Override
  public CompletableFuture<Void> close() {
    raft.deleteObserver(this);
    unregisterHandlers();
    return members.close();
  }

  @Override
  public boolean isClosed() {
    return members.isClosed();
  }

}<commit_msg>Run registerHandlers after members have been created.<commit_after>/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed 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 net.kuujo.copycat.cluster.internal;

import net.kuujo.copycat.EventListener;
import net.kuujo.copycat.cluster.*;
import net.kuujo.copycat.protocol.Protocol;
import net.kuujo.copycat.raft.RaftContext;
import net.kuujo.copycat.raft.protocol.Request;
import net.kuujo.copycat.raft.protocol.Response;
import net.kuujo.copycat.resource.ResourceContext;
import net.kuujo.copycat.util.Managed;
import net.kuujo.copycat.util.concurrent.Futures;
import net.kuujo.copycat.util.internal.Assert;
import net.kuujo.copycat.util.serializer.KryoSerializer;
import net.kuujo.copycat.util.serializer.Serializer;

import java.nio.ByteBuffer;
import java.util.Observable;
import java.util.Observer;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * Default cluster implementation.
 *
 * @author <a href="http://github.com/kuujo"><NAME></a>
 */
public class ManagedCluster implements Cluster, Managed<Cluster>, Observer {
  private final RaftContext raft;
  private final ManagedMembers members;
  private final Set<EventListener<ElectionEvent>> listeners = new CopyOnWriteArraySet<>();
  private final Serializer serializer = new KryoSerializer();
  private String lastLeader;
  private long lastTerm;

  public ManagedCluster(Protocol protocol, ResourceContext context) {
    this.raft = Assert.isNotNull(context, "context").raft();
    this.members = new ManagedMembers(protocol, context);
  }

  @Override
  public void update(Observable o, Object arg) {
    RaftContext raft = (RaftContext) o;
    if (raft.getTerm() > lastTerm && (lastLeader == null || (raft.getLeader() != null && !lastLeader.equals(raft.getLeader())))) {
      lastTerm = raft.getTerm();
      lastLeader = raft.getLeader();
      listeners.forEach(l -> l.accept(new ElectionEvent(ElectionEvent.Type.COMPLETE, lastTerm, member(lastLeader))));
    }
  }

  @Override
  public Member leader() {
    return members.members.get(raft.getLeader());
  }

  @Override
  public long term() {
    return raft.getTerm();
  }

  @Override
  public ManagedLocalMember member() {
    return (ManagedLocalMember) members.members.get(raft.getLocalMember().uri());
  }

  @Override
  public Member member(String uri) {
    return members.members.get(uri);
  }

  @Override
  public Members members() {
    return members;
  }

  @Override
  public <T> Cluster broadcast(String topic, T message) {
    members.forEach(m -> {
      if (m instanceof ManagedRemoteMember) {
        m.send(topic, message);
      }
    });
    return this;
  }

  @Override
  public Cluster addMembershipListener(EventListener<MembershipEvent> listener) {
    members.addListener(listener);
    return this;
  }

  @Override
  public Cluster removeMembershipListener(EventListener<MembershipEvent> listener) {
    members.removeListener(listener);
    return this;
  }

  @Override
  public Cluster addElectionListener(EventListener<ElectionEvent> listener) {
    listeners.add(listener);
    return this;
  }

  @Override
  public Cluster removeElectionListener(EventListener<ElectionEvent> listener) {
    listeners.remove(listener);
    return this;
  }

  /**
   * Registers handlers for the context.
   */
  private void registerHandlers() {
    member().registerInternalHandler(InternalTopics.JOIN, wrapInboundRequest(raft::join));
    member().registerInternalHandler(InternalTopics.PROMOTE, wrapInboundRequest(raft::promote));
    member().registerInternalHandler(InternalTopics.LEAVE, wrapInboundRequest(raft::leave));
    member().registerInternalHandler(InternalTopics.SYNC, wrapInboundRequest(raft::sync));
    member().registerInternalHandler(InternalTopics.POLL, wrapInboundRequest(raft::poll));
    member().registerInternalHandler(InternalTopics.VOTE, wrapInboundRequest(raft::vote));
    member().registerInternalHandler(InternalTopics.APPEND, wrapInboundRequest(raft::append));
    member().registerInternalHandler(InternalTopics.QUERY, wrapInboundRequest(raft::query));
    member().registerInternalHandler(InternalTopics.COMMAND, wrapInboundRequest(raft::command));
    raft.joinHandler(request -> wrapOutboundRequest(InternalTopics.JOIN, request));
    raft.promoteHandler(request -> wrapOutboundRequest(InternalTopics.PROMOTE, request));
    raft.leaveHandler(request -> wrapOutboundRequest(InternalTopics.LEAVE, request));
    raft.syncHandler(request -> wrapOutboundRequest(InternalTopics.SYNC, request));
    raft.pollHandler(request -> wrapOutboundRequest(InternalTopics.POLL, request));
    raft.voteHandler(request -> wrapOutboundRequest(InternalTopics.VOTE, request));
    raft.appendHandler(request -> wrapOutboundRequest(InternalTopics.APPEND, request));
    raft.queryHandler(request -> wrapOutboundRequest(InternalTopics.QUERY, request));
    raft.commandHandler(request -> wrapOutboundRequest(InternalTopics.COMMAND, request));
  }

  /**
   * Unregisters handlers for the context.
   */
  private void unregisterHandlers() {
    member().unregisterInternalHandler(InternalTopics.JOIN);
    member().unregisterInternalHandler(InternalTopics.PROMOTE);
    member().unregisterInternalHandler(InternalTopics.LEAVE);
    member().unregisterInternalHandler(InternalTopics.SYNC);
    member().unregisterInternalHandler(InternalTopics.POLL);
    member().unregisterInternalHandler(InternalTopics.VOTE);
    member().unregisterInternalHandler(InternalTopics.APPEND);
    member().unregisterInternalHandler(InternalTopics.QUERY);
    member().unregisterInternalHandler(InternalTopics.COMMAND);
    raft.joinHandler(null);
    raft.promoteHandler(null);
    raft.leaveHandler(null);
    raft.syncHandler(null);
    raft.pollHandler(null);
    raft.voteHandler(null);
    raft.appendHandler(null);
    raft.queryHandler(null);
    raft.commandHandler(null);
  }

  /**
   * Wraps an inbound context call.
   */
  private <T, U> MessageHandler<ByteBuffer, ByteBuffer> wrapInboundRequest(MessageHandler<T, U> handler) {
    return message -> CompletableFuture.completedFuture(null)
        .thenCompose(v -> handler.apply(serializer.readObject(message)))
        .thenApply(serializer::writeObject);
  }

  /**
   * Wraps an outbound context call.
   */
  private <T extends Request, U extends Response> CompletableFuture<U> wrapOutboundRequest(String topic, T request) {
    ManagedMember<?> member = members.members.get(request.uri());
    if (member != null) {
      return member.sendInternal(topic, serializer.writeObject(request)).thenApply(serializer::readObject);
    }
    return Futures.exceptionalFuture(new ClusterException("Unknown member URI"));
  }

  @Override
  public CompletableFuture<Cluster> open() {
    raft.addObserver(this);
    return members.open()
      .thenRun(this::registerHandlers)
      .thenApply(v -> this);
  }

  @Override
  public boolean isOpen() {
    return members.isOpen();
  }

  @Override
  public CompletableFuture<Void> close() {
    raft.deleteObserver(this);
    unregisterHandlers();
    return members.close();
  }

  @Override
  public boolean isClosed() {
    return members.isClosed();
  }

}
{'fraction_non_alphanumeric': 0.11466165413533834, 'fraction_numerical': 0.0010370754472387865, 'mean_word_length': 4.809111445783133, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '680948', 'n_tokens_mistral': 4484, 'n_tokens_neox': 4370, 'n_words': 891}