package org.mitallast.queue.raft.state;

import com.google.common.collect.ImmutableList;
import com.google.common.net.HostAndPort;
import com.google.inject.Inject;
import org.mitallast.queue.Version;
import org.mitallast.queue.common.component.AbstractLifecycleComponent;
import org.mitallast.queue.common.settings.Settings;
import org.mitallast.queue.transport.DiscoveryNode;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class ClusterState extends AbstractLifecycleComponent implements Iterable<MemberState> {
    private final Map<DiscoveryNode, MemberState> members = new HashMap<>();
    private final List<MemberState> activeMembers = new ArrayList<>();
    private final List<MemberState> passiveMembers = new ArrayList<>();

    @Inject
    public ClusterState(Settings settings) {
        super(settings);
    }

    public ClusterState addMember(MemberState member) {
        assert members.putIfAbsent(member.getNode(), member) == null;
        if (member.getType() == MemberState.Type.ACTIVE) {
            addActiveMember(member);
        } else {
            addPassiveMember(member);
        }
        return this;
    }

    private void addActiveMember(MemberState member) {
        activeMembers.add(member);
        logger.info("add active member {}", member.getNode());
        logger.info("active members: {}", activeMembers);
    }

    private void addPassiveMember(MemberState member) {
        passiveMembers.add(member);
    }

    ClusterState removeMember(MemberState member) {
        members.remove(member.getNode());
        if (member.getType() == MemberState.Type.ACTIVE) {
            removeActiveMember(member);
        } else {
            removePassiveMember(member);
        }
        return this;
    }

    private void removeActiveMember(MemberState member) {
        Iterator<MemberState> iterator = activeMembers.iterator();
        while (iterator.hasNext()) {
            if (iterator.next().getNode().equals(member.getNode())) {
                iterator.remove();
            }
        }
    }

    private void removePassiveMember(MemberState member) {
        Iterator<MemberState> iterator = passiveMembers.iterator();
        while (iterator.hasNext()) {
            if (iterator.next().getNode().equals(member.getNode())) {
                iterator.remove();
            }
        }
    }

    public MemberState getMember(DiscoveryNode node) {
        return members.get(node);
    }

    public ImmutableList<MemberState> getActiveMembers() {
        return ImmutableList.copyOf(activeMembers);
    }

    public ImmutableList<MemberState> getPassiveMembers() {
        return ImmutableList.copyOf(passiveMembers);
    }

    public ImmutableList<MemberState> getMembers() {
        return ImmutableList.copyOf(members.values());
    }

    public ImmutableList<DiscoveryNode> activeNodes() {
        return ImmutableList.copyOf(getActiveMembers().stream().map(MemberState::getNode).collect(Collectors.toList()));
    }

    public ImmutableList<DiscoveryNode> passiveNodes() {
        return ImmutableList.copyOf(getPassiveMembers().stream().map(MemberState::getNode).collect(Collectors.toList()));
    }

    public ImmutableList<DiscoveryNode> nodes() {
        return ImmutableList.copyOf(getMembers().stream().map(MemberState::getNode).collect(Collectors.toList()));
    }

    @Override
    public Iterator<MemberState> iterator() {
        return new ClusterStateIterator(members.entrySet().iterator());
    }

    @Override
    protected void doStart() throws IOException {
        String[] nodes = settings.getAsArray("raft.cluster.nodes");
        for (String node : nodes) {
            HostAndPort hostAndPort = HostAndPort.fromString(node);
            DiscoveryNode discoveryNode = new DiscoveryNode(node, hostAndPort, Version.CURRENT);
            addMember(new MemberState(discoveryNode, MemberState.Type.ACTIVE));
        }
    }

    @Override
    protected void doStop() throws IOException {

    }

    @Override
    protected void doClose() throws IOException {

    }

    private class ClusterStateIterator implements Iterator<MemberState> {
        private final Iterator<Map.Entry<DiscoveryNode, MemberState>> iterator;
        private MemberState member;

        private ClusterStateIterator(Iterator<Map.Entry<DiscoveryNode, MemberState>> iterator) {
            this.iterator = iterator;
        }

        @Override
        public boolean hasNext() {
            return iterator.hasNext();
        }

        @Override
        public MemberState next() {
            member = iterator.next().getValue();
            return member;
        }

        @Override
        public void remove() {
            iterator.remove();
            removeMember(member);
        }
    }
}

@@ -369,16 +369,71 @@
 eryNode;
+%0Aimport org.mitallast.queue.transport.TransportService;
 %0A%0Aimport
@@ -826,16 +826,69 @@
 ist%3C%3E();
+%0A    private final TransportService transportService;
 %0A%0A    @I
@@ -938,35 +938,120 @@
 ings
-) %7B%0A        super(settings)
+, TransportService transportService) %7B%0A        super(settings);%0A        this.transportService = transportService
 ;%0A  
@@ -1123,15 +1123,12 @@
     
-assert 
+if (
 memb
@@ -1176,18 +1176,24 @@
  == null
-;%0A
+) %7B%0A    
         
@@ -1235,32 +1235,36 @@
 .Type.ACTIVE) %7B%0A
+    
             addA
@@ -1284,32 +1284,36 @@
 ember);%0A        
+    
 %7D else %7B%0A       
@@ -1297,32 +1297,36 @@
        %7D else %7B%0A
+    
             addP
@@ -1339,32 +1339,46 @@
 Member(member);%0A
+            %7D%0A
         %7D%0A      
@@ -3797,16 +3797,219 @@
 ption %7B%0A
+        MemberState.Type type = settings.getAsBoolean(%22raft.passive%22, false) ? MemberState.Type.PASSIVE : MemberState.Type.ACTIVE;%0A        addMember(new MemberState(transportService.localNode(), type));%0A
         
@@ -4168,24 +4168,100 @@
 ring(node);%0A
+            if (!hostAndPort.equals(transportService.localAddress())) %7B%0A    
             
@@ -4349,32 +4349,36 @@
 T);%0A            
+    
 addMember(new Me
@@ -4425,24 +4425,38 @@
 e.ACTIVE));%0A
+            %7D%0A
         %7D%0A  

fix cluster state