<commit_before>package org.jgroups.tests;

import org.jgroups.*;
import org.jgroups.util.Util;

import java.security.SecureRandom;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Verify that all messages are delivered in the view they are sent in
 * regardless of members joining, leaving or crashing.
 * @author rnewson
 *
 */
public final class ViewDeliveryDemo {


    private static final int SEND = 0;
    private static final int REOPEN = 1;
    private static final int RECONNECT = 2;

    private static Channel channel=null;
    private static final Random random = new SecureRandom();
    private static MyReceiver mr = null;

    static String props="flush-udp.xml";

    public static void main(final String[] args) throws Exception {
        for(int i=0; i < args.length; i++) {
            if(args[i].equals("-props")) {
                props=args[++i];
                continue;
            }
            System.out.println("ViewDeliveryDemo [-help] [-props <props>]");
            return;
        }

        channel=new JChannel(props);
        channel.connect("view_test");
        mr = new MyReceiver();
        channel.setReceiver(mr);

        while (true) {
            switch (random.nextInt(3)) {
            case SEND:
                send();
                break;
            case REOPEN:
                reopen();
                break;
            case RECONNECT:
                reconnect();
                break;
            default:
                assert false;
            }
            Thread.sleep(random.nextInt(2000));
        }
    }

    private static void send() throws Exception {
        int max=random.nextInt(1000);
        System.out.println("Sending " + max + " messages");
        for (int i = 0; i < max; i++) {
            channel.send(null, null, mr.getViewId());
        }
    }

    private static void reopen() throws Exception {
        System.out.println("closing and reopening.");
        channel.close();
        Thread.sleep(random.nextInt(5000));
        channel.open();
        channel.connect("view_test");
    }

    private static void reconnect() throws Exception {
        System.out.println("disconnecting and reconnecting.");
        channel.disconnect();
        Thread.sleep(random.nextInt(5000));
        channel.connect("view_test");
    }



    private static class MyReceiver extends ReceiverAdapter implements Runnable {
        ViewId my_vid;
        long last_time=System.currentTimeMillis();
        static final long MAX_TIME=10000;
        final AtomicInteger count=new AtomicInteger(0), violations=new AtomicInteger(0);

        private MyReceiver() {
            new Thread(this).start();
        }

        public void run() {
            while(true) {
                Util.sleep(MAX_TIME);
                System.out.println("==> received " + count.get() + " valid msgs, " + violations.get() + " violations so far");
            }
        }


        public void viewAccepted(View new_view) {
            System.out.println("new_view = " + new_view);
            my_vid=new_view.getVid();
        }
        
        public ViewId getViewId(){
            return my_vid;            
        }

        public void receive(final Message msg) {
            final Object obj = msg.getObject ();
            if (obj instanceof ViewId) {
                count.incrementAndGet();
                final ViewId sent_in_vid = (ViewId) obj;
                final ViewId arrived_in_vid = my_vid;
                if (!sent_in_vid.equals(arrived_in_vid)) {
                    System.out.printf("******** VIOLATION: message sent in view %s received in %s\n",
                                      sent_in_vid, arrived_in_vid);
                    violations.incrementAndGet();
                }
            }
            else {
                System.out.println("ERROR: unexpected payload: " + obj);
            }
        }
    }

<commit_msg>added block() and unblock() support<commit_after>package org.jgroups.tests;

import org.jgroups.*;
import org.jgroups.util.Util;

import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Verify that all messages are delivered in the view they are sent in
 * regardless of members joining, leaving or crashing.
 * @author rnewson
 *
 */
public final class ViewDeliveryDemo {


    private static final int SEND = 0;
    private static final int REOPEN = 1;
    private static final int RECONNECT = 2;

    private static Channel channel=null;
    private static final Lock lock=new ReentrantLock();
    private static boolean blocked=false;
    private static final Random random = new Random();
    private static MyReceiver mr = null;

    static String props="flush-udp.xml";

    public static void main(final String[] args) throws Exception {
        for(int i=0; i < args.length; i++) {
            if(args[i].equals("-props")) {
                props=args[++i];
                continue;
            }
            System.out.println("ViewDeliveryDemo [-help] [-props <props>]");
            return;
        }

        channel=new JChannel(props);
        channel.setOpt(Channel.BLOCK, true);
        channel.connect("view_test");
        mr = new MyReceiver();
        channel.setReceiver(mr);

        while (true) {
            switch (random.nextInt(3)) {
            case SEND:
                lock.lock();
                try {
                    if(!blocked)
                        send();
                    else
                        System.out.println("Didn't send any messages because I was blocked");
                }
                finally {
                    lock.unlock();
                }

                break;
            case REOPEN:
                reopen();
                break;
            case RECONNECT:
                reconnect();
                break;
            default:
                assert false;
            }
            Thread.sleep(random.nextInt(2000));
        }
    }

    private static void send() throws Exception {
        int max=random.nextInt(1000);
        System.out.println("Sending " + max + " messages");
        for (int i = 0; i < max; i++) {
            channel.send(null, null, channel.getView().getVid());
        }
    }

    private static void reopen() throws Exception {
        System.out.println("closing and reopening.");
        channel.close();
        Thread.sleep(random.nextInt(5000));
        channel.open();
        channel.connect("view_test");
    }

    private static void reconnect() throws Exception {
        System.out.println("disconnecting and reconnecting.");
        channel.disconnect();
        Thread.sleep(random.nextInt(5000));
        channel.connect("view_test");
    }



    private static class MyReceiver extends ExtendedReceiverAdapter implements Runnable {
        ViewId my_vid;
        long last_time=System.currentTimeMillis();
        static final long MAX_TIME=10000;
        final AtomicInteger count=new AtomicInteger(0), violations=new AtomicInteger(0);

        private MyReceiver() {
            new Thread(this).start();
        }

        public void run() {
            while(true) {
                Util.sleep(MAX_TIME);
                System.out.println("==> received " + count.get() + " valid msgs, " + violations.get() + " violations so far");
            }
        }


        public void viewAccepted(View new_view) {
            System.out.println("new_view = " + new_view);
            my_vid=new_view.getVid();
        }
        

        public void receive(final Message msg) {
            final Object obj = msg.getObject ();
            if (obj instanceof ViewId) {
                count.incrementAndGet();
                final ViewId sent_in_vid = (ViewId) obj;
                final ViewId arrived_in_vid = my_vid;
                if (!sent_in_vid.equals(arrived_in_vid)) {
                    System.out.printf("******** VIOLATION: message sent in view %s received in %s\n",
                                      sent_in_vid, arrived_in_vid);
                    violations.incrementAndGet();
                }
            }
            else {
                System.out.println("ERROR: unexpected payload: " + obj);
            }
        }

        public void block() {
            System.out.println("block()");
            lock.lock();
            try {
                blocked=true;
            }
            finally {
                lock.unlock();
            }
        }

        public void unblock() {
            System.out.println("unblock()");
            lock.lock();
            try {
                blocked=false;
            }
            finally {
                lock.unlock();
            }
        }
    }


{'fraction_non_alphanumeric': 0.10612198964663516, 'fraction_numerical': 0.006527121314427189, 'mean_word_length': 1.6599820413049986, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6810593', 'n_tokens_mistral': 2425, 'n_tokens_neox': 2273, 'n_words': 575}