<commit_before>package com.aol.simple.react.simple;

import static com.aol.simple.react.stream.simple.SimpleReact.iterate;
import static com.aol.simple.react.stream.simple.SimpleReact.times;
import static com.aol.simple.react.stream.simple.SimpleReact.timesInSequence;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertThat;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;

import org.junit.Test;

import com.aol.simple.react.stream.simple.SimpleReact;
import com.aol.simple.react.stream.traits.SimpleReactStream;

public class GeneratorTest {
	volatile int  count;
	volatile int  second;
	volatile int  capture;
	
	private Object lock1= "lock1";
	private Object lock2= "lock2";
	
	public synchronized void incrementCapture(){
		capture++;
	}
	public synchronized int incrementCount(){
		return count++;
	}
	@Test
	public void testGenerate() throws InterruptedException, ExecutionException {
		count =0;
		capture =0;
		List<String> strings = new SimpleReact()
				.<Integer> react(() -> incrementCount() ,SimpleReact.times(4))
				.then(it -> it * 100)
				.then(it -> "*" + it)
				.capture(e -> incrementCapture())
				.block();

		assertThat("Capture is " +  capture,strings.size(), is(4));
		assertThat("Capture is " +  capture,count,is(4));
	}
	
	@Test
	public void testGenerateOffset() throws InterruptedException, ExecutionException {
		count =0;
		
		List<String> strings = new SimpleReact()
				.<Integer> react(() -> incrementCount() ,timesInSequence(1).offset(2))
				.then(it -> it * 100)
				.then(it -> "*" + it)
				.block();

		System.out.println(strings.get(0));
		System.out.println(count);
		assertThat(strings.size(), is(1));
		
		assertThat(count,is(3)); 
				
		
	}
	@Test
	public void testGenerateDataflowMovingConcurrently() throws InterruptedException, ExecutionException {
		AtomicInteger count =new AtomicInteger();
		AtomicInteger second = new AtomicInteger();
		SimpleReactStream s = new SimpleReact()
				.<Integer> react(() -> {
					sleep(count.incrementAndGet());
					return count.get();
				} ,times(50))
				.then(it -> it * 100)
				.then(it -> {
					
					second.incrementAndGet();
					return it;
				})
				.<String>then(it -> "*" + it);<commit_msg>add ignore for non-deterministic tests for CI<commit_after>package com.aol.simple.react.simple;

import static com.aol.simple.react.stream.simple.SimpleReact.iterate;
import static com.aol.simple.react.stream.simple.SimpleReact.times;
import static com.aol.simple.react.stream.simple.SimpleReact.timesInSequence;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertThat;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;

import org.junit.Ignore;
import org.junit.Test;

import com.aol.simple.react.stream.simple.SimpleReact;
import com.aol.simple.react.stream.traits.SimpleReactStream;

public class GeneratorTest {
	volatile int  count;
	volatile int  second;
	volatile int  capture;
	
	private Object lock1= "lock1";
	private Object lock2= "lock2";
	
	public synchronized void incrementCapture(){
		capture++;
	}
	public synchronized int incrementCount(){
		return count++;
	}
	@Test
	public void testGenerate() throws InterruptedException, ExecutionException {
		count =0;
		capture =0;
		List<String> strings = new SimpleReact()
				.<Integer> react(() -> incrementCount() ,SimpleReact.times(4))
				.then(it -> it * 100)
				.then(it -> "*" + it)
				.capture(e -> incrementCapture())
				.block();

		assertThat("Capture is " +  capture,strings.size(), is(4));
		assertThat("Capture is " +  capture,count,is(4));
	}
	
	@Test
	public void testGenerateOffset() throws InterruptedException, ExecutionException {
		count =0;
		
		List<String> strings = new SimpleReact()
				.<Integer> react(() -> incrementCount() ,timesInSequence(1).offset(2))
				.then(it -> it * 100)
				.then(it -> "*" + it)
				.block();

		System.out.println(strings.get(0));
		System.out.println(count);
		assertThat(strings.size(), is(1));
		
		assertThat(count,is(3)); 
				
		
	}
	@Test @Ignore //non-deterministic
	public void testGenerateDataflowMovingConcurrently() throws InterruptedException, ExecutionException {
		AtomicInteger count =new AtomicInteger();
		AtomicInteger second = new AtomicInteger();
		SimpleReactStream s = new SimpleReact()
				.<Integer> react(() -> {
					sleep(count.incrementAndGet());
					return count.get();
				} ,times(50))
				.then(it -> it * 100)
				.then(it -> {
					
					second.incrementAndGet();
					return it;
				})
				.<String>then(it -> "*" + it);
{'fraction_non_alphanumeric': 0.13481453586895123, 'fraction_numerical': 0.009790999811711542, 'mean_word_length': 4.254203758654797, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 37, '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': '2767245', 'n_tokens_mistral': 1882, 'n_tokens_neox': 1654, 'n_words': 361}