<commit_before>package com.jivesoftware.os.aquarium;

import com.jivesoftware.os.aquarium.interfaces.AtQuorum;
import com.jivesoftware.os.aquarium.interfaces.StateStorage;
import com.jivesoftware.os.aquarium.interfaces.StreamQuorumState;
import com.jivesoftware.os.aquarium.interfaces.MemberLifecycle;
import com.google.common.collect.Sets;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.lang.reflect.Array;
import java.util.Set;

/**
 * @author jonathan.colt
 */
public class ReadWaterline<T> {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final StateStorage<T> stateStorage;
    private final MemberLifecycle<T> memberLifecycle;
    private final AtQuorum atQuorum;
    private final Class<T> lifecycleType;

    public ReadWaterline(StateStorage<T> stateStorage,
        MemberLifecycle<T> memberLifecycle,
        AtQuorum atQuorum,
        Class<T> lifecycleType) {
        this.stateStorage = stateStorage;
        this.memberLifecycle = memberLifecycle;
        this.atQuorum = atQuorum;
        this.lifecycleType = lifecycleType;
    }

    public Waterline get(Member asMember) throws Exception {
        TimestampedState[] current = new TimestampedState[1];
        Set<Member> acked = Sets.newHashSet();
        T lifecycle = memberLifecycle.get(asMember);
        if (lifecycle == null) {
            //LOG.info("Null lifecycle for {}", asMember);
            return null;
        }
        stateStorage.scan(asMember, null, lifecycle, (rootRingMember, isSelf, ackRingMember, rootLifecycle, state, timestamp, version) -> {
            if (current[0] == null && isSelf) {
                current[0] = new TimestampedState(state, timestamp, version);
            }
            if (current[0] != null) {
                TimestampedState v = new TimestampedState(state, timestamp, version);
                if (v.state == current[0].state && v.timestamp == current[0].timestamp) {
                    acked.add(ackRingMember);
                }
            }
            return true;
        });
        if (current[0] != null) {
            return new Waterline(asMember,
                current[0].state,
                current[0].timestamp,
                current[0].version,
                atQuorum.is(acked.size()));
        } else {
            return null;
        }
    }

    public void getOthers(Member asMember, StreamQuorumState stream) throws Exception {
        Member[] otherMember = new Member[1];
        TimestampedState[] otherState = new TimestampedState[1];
        @SuppressWarnings("unchecked")
        T[] otherLifecycle = (T[]) Array.newInstance(lifecycleType, 1);
        Set<Member> acked = Sets.newHashSet();
        stateStorage.scan(null, null, null, (rootMember, isSelf, ackMember, rootLifecycle, state, timestamp, version) -> {
            if (otherMember[0] != null && !otherMember[0].equals(rootMember)) {
                boolean otherHasQuorum = atQuorum.is(acked.size());
                stream.stream(new Waterline(otherMember[0],
                    otherState[0].state,
                    otherState[0].timestamp,
                    otherState[0].version,
                    otherHasQuorum));

                otherMember[0] = null;
                otherState[0] = null;
                acked.clear();
            }

            if (otherMember[0] == null && isSelf && !asMember.equals(rootMember)) {
                T lifecycle = memberLifecycle.get(rootMember);
                if (lifecycle != null) {
                    otherMember[0] = rootMember;
                    otherState[0] = new TimestampedState(state, timestamp, version);
                    otherLifecycle[0] = lifecycle;
                }
            }
            if (otherMember[0] != null) {
                TimestampedState v = new TimestampedState(state, timestamp, version);
                if (otherLifecycle[0].equals(rootLifecycle) && v.state == otherState[0].state && v.timestamp == otherState[0].timestamp) {
                    acked.add(ackMember);
                }
            }
            return true;
        });

        if (otherMember[0] != null) {
            boolean otherHasQuorum = atQuorum.is(acked.size());
            stream.stream(new Waterline(otherMember[0],
                otherState[0].state,
                otherState[0].timestamp,
                otherState[0].version,
                otherHasQuorum));
        }
    }

    public void acknowledgeOther(Member member) throws Exception {
        stateStorage.update(setState -> {
            @SuppressWarnings("unchecked")<commit_msg>properly handle multiple lifecycles for other members<commit_after>package com.jivesoftware.os.aquarium;

import com.jivesoftware.os.aquarium.interfaces.AtQuorum;
import com.jivesoftware.os.aquarium.interfaces.StateStorage;
import com.jivesoftware.os.aquarium.interfaces.StreamQuorumState;
import com.jivesoftware.os.aquarium.interfaces.MemberLifecycle;
import com.google.common.collect.Sets;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.lang.reflect.Array;
import java.util.Set;

/**
 * @author jonathan.colt
 */
public class ReadWaterline<T> {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final StateStorage<T> stateStorage;
    private final MemberLifecycle<T> memberLifecycle;
    private final AtQuorum atQuorum;
    private final Class<T> lifecycleType;

    public ReadWaterline(StateStorage<T> stateStorage,
        MemberLifecycle<T> memberLifecycle,
        AtQuorum atQuorum,
        Class<T> lifecycleType) {
        this.stateStorage = stateStorage;
        this.memberLifecycle = memberLifecycle;
        this.atQuorum = atQuorum;
        this.lifecycleType = lifecycleType;
    }

    public Waterline get(Member asMember) throws Exception {
        TimestampedState[] current = new TimestampedState[1];
        Set<Member> acked = Sets.newHashSet();
        T lifecycle = memberLifecycle.get(asMember);
        if (lifecycle == null) {
            //LOG.info("Null lifecycle for {}", asMember);
            return null;
        }
        stateStorage.scan(asMember, null, lifecycle, (rootRingMember, isSelf, ackRingMember, rootLifecycle, state, timestamp, version) -> {
            if (current[0] == null && isSelf) {
                current[0] = new TimestampedState(state, timestamp, version);
            }
            if (current[0] != null) {
                TimestampedState v = new TimestampedState(state, timestamp, version);
                if (v.state == current[0].state && v.timestamp == current[0].timestamp) {
                    acked.add(ackRingMember);
                }
            }
            return true;
        });
        if (current[0] != null) {
            return new Waterline(asMember,
                current[0].state,
                current[0].timestamp,
                current[0].version,
                atQuorum.is(acked.size()));
        } else {
            return null;
        }
    }

    public void getOthers(Member asMember, StreamQuorumState stream) throws Exception {
        Member[] otherMember = new Member[1];
        TimestampedState[] otherState = new TimestampedState[1];
        @SuppressWarnings("unchecked")
        T[] otherLifecycle = (T[]) Array.newInstance(lifecycleType, 1);
        Set<Member> acked = Sets.newHashSet();
        stateStorage.scan(null, null, null, (rootMember, isSelf, ackMember, rootLifecycle, state, timestamp, version) -> {
            if (otherMember[0] != null && !otherMember[0].equals(rootMember)) {
                boolean otherHasQuorum = atQuorum.is(acked.size());
                stream.stream(new Waterline(otherMember[0],
                    otherState[0].state,
                    otherState[0].timestamp,
                    otherState[0].version,
                    otherHasQuorum));

                otherMember[0] = null;
                otherState[0] = null;
                acked.clear();
            }

            if (otherMember[0] == null && isSelf && !asMember.equals(rootMember)) {
                T lifecycle = memberLifecycle.get(rootMember);
                if (lifecycle != null && rootLifecycle.equals(lifecycle)) {
                    otherMember[0] = rootMember;
                    otherState[0] = new TimestampedState(state, timestamp, version);
                    otherLifecycle[0] = lifecycle;
                }
            }
            if (otherMember[0] != null) {
                TimestampedState v = new TimestampedState(state, timestamp, version);
                if (otherLifecycle[0].equals(rootLifecycle) && v.state == otherState[0].state && v.timestamp == otherState[0].timestamp) {
                    acked.add(ackMember);
                }
            }
            return true;
        });

        if (otherMember[0] != null) {
            boolean otherHasQuorum = atQuorum.is(acked.size());
            stream.stream(new Waterline(otherMember[0],
                otherState[0].state,
                otherState[0].timestamp,
                otherState[0].version,
                otherHasQuorum));
        }
    }

    public void acknowledgeOther(Member member) throws Exception {
        stateStorage.update(setState -> {
            @SuppressWarnings("unchecked")
{'fraction_non_alphanumeric': 0.10820855898906233, 'fraction_numerical': 0.007220983328023786, 'mean_word_length': 2.1646505376344085, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3836280', 'n_tokens_mistral': 2637, 'n_tokens_neox': 2539, 'n_words': 504}