<commit_before>package com.orbekk.same;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.orbekk.same.State.Component;

public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private ConnectionManager connections;
    private State state;
    private boolean stopped = false;
    private Broadcaster broadcaster;
    private List<String> _fullStateReceivers = new ArrayList<String>();
    
    public MasterServiceImpl(State initialState, ConnectionManager connections,
            Broadcaster broadcaster) {
        state = initialState;
        this.broadcaster = broadcaster;
    }
    
    @Override
    public void joinNetworkRequest(String clientUrl) {
        logger.info("JoinNetworkRequest({})", clientUrl);
        List<String> participants = participants();
        if (!participants.contains(clientUrl)) {
            participants.add(clientUrl);
            synchronized(this) {
                notifyAll();
                state.updateFromObject(".participants", participants,
                        state.getRevision(".participants"));
            }
        } else {                
            logger.warn("Client {} already part of network. " +
                    "Ignoring participation request", clientUrl);
        }
    }
    
    public boolean _sendUpdatedComponents() {
        boolean worked = false;
        for (final String component : state.getAndClearUpdatedComponents()) {
            logger.info("Broadcasting new component {}", state.show(component));
            broadcaster.broadcast(participants(), new ServiceOperation() {
               @Override public void run(ClientService client) {
                   client.setState(component, state.getDataOf(component),
                           state.getRevision(component));
               }
            });
            worked = true;
        }
        return worked;
    }
    
    public synchronized boolean _sendFullState() {
        boolean hasWork = _fullStateReceivers.size() != 0;
        if (hasWork) {
            final List<State.Component> components = state.getComponents();
            broadcaster.broadcast(participants(), new ServiceOperation() {
                @Override public void run(ClientService client) {
                    for (Component c : components) {
                        client.setState(c.getName(), c.getData(), c.getRevision());
                    }
                }
            });
            _fullStateReceivers.clear();
        }
        return hasWork;
    }
    
    private List<String> participants() {
        return state.getList(".participants");
    }
    
    
    @Override
    public synchronized boolean updateStateRequest(String component,
            String newData, long revision) {
        boolean updated = state.update(component, newData, revision);
        if (updated) {
            notifyAll();<commit_msg>Fix full state update to new clients.<commit_after>package com.orbekk.same;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.orbekk.same.State.Component;

public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private ConnectionManager connections;
    private State state;
    private boolean stopped = false;
    private Broadcaster broadcaster;
    private List<String> _fullStateReceivers = new ArrayList<String>();
    
    public MasterServiceImpl(State initialState, ConnectionManager connections,
            Broadcaster broadcaster) {
        state = initialState;
        this.broadcaster = broadcaster;
    }
    
    @Override
    public void joinNetworkRequest(String clientUrl) {
        logger.info("JoinNetworkRequest({})", clientUrl);
        List<String> participants = participants();
        if (!participants.contains(clientUrl)) {
            participants.add(clientUrl);
            _fullStateReceivers.add(clientUrl);
            synchronized(this) {
                state.updateFromObject(".participants", participants,
                        state.getRevision(".participants"));
                notifyAll();
            }
        } else {                
            logger.warn("Client {} already part of network. " +
                    "Ignoring participation request", clientUrl);
        }
    }
    
    public boolean _sendUpdatedComponents() {
        boolean worked = false;
        for (final String component : state.getAndClearUpdatedComponents()) {
            logger.info("Broadcasting new component {}", state.show(component));
            broadcaster.broadcast(participants(), new ServiceOperation() {
               @Override public void run(ClientService client) {
                   client.setState(component, state.getDataOf(component),
                           state.getRevision(component));
               }
            });
            worked = true;
        }
        return worked;
    }
    
    public synchronized boolean _sendFullState() {
        boolean hasWork = _fullStateReceivers.size() != 0;
        if (hasWork) {
            final List<State.Component> components = state.getComponents();
            broadcaster.broadcast(_fullStateReceivers, new ServiceOperation() {
                @Override public void run(ClientService client) {
                    for (Component c : components) {
                        client.setState(c.getName(), c.getData(), c.getRevision());
                    }
                }
            });
            _fullStateReceivers.clear();
        }
        return hasWork;
    }
    
    private List<String> participants() {
        return state.getList(".participants");
    }
    
    
    @Override
    public synchronized boolean updateStateRequest(String component,
            String newData, long revision) {
        boolean updated = state.update(component, newData, revision);
        if (updated) {
            notifyAll();
{'fraction_non_alphanumeric': 0.09233534891322112, 'fraction_numerical': 0.0009805523778395162, 'mean_word_length': 2.144912641315519, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '933912', 'n_tokens_mistral': 1481, 'n_tokens_neox': 1439, 'n_words': 336}