JAVA-12045 : move java concurrency ebook content code to common module (#12559)

* JAVA-12045 : move java concurrency ebook content code to common module * JAVA-12045: addressed the review comments .. renamed package names and placed core-java-concurrency-simple module in pom.xml * JAVA-12045: removed duplicate module after renaming java-simple-module to core-java-simple-module
2022-08-08 23:26:55 +10:00
parent 9528c51654
commit cdb19ff6d5
36 changed files with 45 additions and 8 deletions
@@ -0,0 +1,208 @@
+package com.baeldung.completablefuture;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class CompletableFutureLongRunningUnitTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CompletableFutureLongRunningUnitTest.class);
+
+    @Test
+    public void whenRunningCompletableFutureAsynchronously_thenGetMethodWaitsForResult() throws InterruptedException, ExecutionException {
+        Future<String> completableFuture = calculateAsync();
+
+        String result = completableFuture.get();
+        assertEquals("Hello", result);
+    }
+
+    private Future<String> calculateAsync() throws InterruptedException {
+        CompletableFuture<String> completableFuture = new CompletableFuture<>();
+
+        Executors.newCachedThreadPool()
+            .submit(() -> {
+                Thread.sleep(500);
+                completableFuture.complete("Hello");
+                return null;
+            });
+
+        return completableFuture;
+    }
+
+    @Test
+    public void whenRunningCompletableFutureWithResult_thenGetMethodReturnsImmediately() throws InterruptedException, ExecutionException {
+        Future<String> completableFuture = CompletableFuture.completedFuture("Hello");
+
+        String result = completableFuture.get();
+        assertEquals("Hello", result);
+    }
+
+    private Future<String> calculateAsyncWithCancellation() throws InterruptedException {
+        CompletableFuture<String> completableFuture = new CompletableFuture<>();
+
+        Executors.newCachedThreadPool()
+            .submit(() -> {
+                Thread.sleep(500);
+                completableFuture.cancel(false);
+                return null;
+            });
+
+        return completableFuture;
+    }
+
+    @Test(expected = CancellationException.class)
+    public void whenCancelingTheFuture_thenThrowsCancellationException() throws ExecutionException, InterruptedException {
+        Future<String> future = calculateAsyncWithCancellation();
+        future.get();
+    }
+
+    @Test
+    public void whenCreatingCompletableFutureWithSupplyAsync_thenFutureReturnsValue() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
+
+        assertEquals("Hello", future.get());
+    }
+
+    @Test
+    public void whenAddingThenAcceptToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
+
+        CompletableFuture<Void> future = completableFuture.thenAccept(s -> LOG.debug("Computation returned: " + s));
+
+        future.get();
+    }
+
+    @Test
+    public void whenAddingThenRunToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
+
+        CompletableFuture<Void> future = completableFuture.thenRun(() -> LOG.debug("Computation finished."));
+
+        future.get();
+    }
+
+    @Test
+    public void whenAddingThenApplyToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
+
+        CompletableFuture<String> future = completableFuture.thenApply(s -> s + " World");
+
+        assertEquals("Hello World", future.get());
+    }
+
+    @Test
+    public void whenUsingThenCompose_thenFuturesExecuteSequentially() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello")
+            .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));
+
+        assertEquals("Hello World", completableFuture.get());
+    }
+
+    @Test
+    public void whenUsingThenCombine_thenWaitForExecutionOfBothFutures() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello")
+            .thenCombine(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> s1 + s2);
+
+        assertEquals("Hello World", completableFuture.get());
+    }
+
+    @Test
+    public void whenUsingThenAcceptBoth_thenWaitForExecutionOfBothFutures() throws ExecutionException, InterruptedException {
+        CompletableFuture.supplyAsync(() -> "Hello")
+            .thenAcceptBoth(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> LOG.debug(s1 + s2));
+    }
+
+    @Test
+    public void whenFutureCombinedWithAllOfCompletes_thenAllFuturesAreDone() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
+        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Beautiful");
+        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "World");
+
+        CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2, future3);
+
+        // ...
+
+        combinedFuture.get();
+
+        assertTrue(future1.isDone());
+        assertTrue(future2.isDone());
+        assertTrue(future3.isDone());
+
+        String combined = Stream.of(future1, future2, future3)
+            .map(CompletableFuture::join)
+            .collect(Collectors.joining(" "));
+
+        assertEquals("Hello Beautiful World", combined);
+    }
+
+    @Test
+    public void whenFutureThrows_thenHandleMethodReceivesException() throws ExecutionException, InterruptedException {
+        String name = null;
+
+        // ...
+
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
+            if (name == null) {
+                throw new RuntimeException("Computation error!");
+            }
+            return "Hello, " + name;
+        })
+            .handle((s, t) -> s != null ? s : "Hello, Stranger!");
+
+        assertEquals("Hello, Stranger!", completableFuture.get());
+    }
+
+    @Test(expected = ExecutionException.class)
+    public void whenCompletingFutureExceptionally_thenGetMethodThrows() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = new CompletableFuture<>();
+
+        // ...
+
+        completableFuture.completeExceptionally(new RuntimeException("Calculation failed!"));
+
+        // ...
+
+        completableFuture.get();
+    }
+
+    @Test
+    public void whenAddingThenApplyAsyncToFuture_thenFunctionExecutesAfterComputationIsFinished() throws ExecutionException, InterruptedException {
+        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
+
+        CompletableFuture<String> future = completableFuture.thenApplyAsync(s -> s + " World");
+
+        assertEquals("Hello World", future.get());
+    }
+    
+    @Test
+    public void whenPassingTransformation_thenFunctionExecutionWithThenApply() throws InterruptedException, ExecutionException {
+        CompletableFuture<Integer> finalResult = compute().thenApply(s -> s + 1);
+        assertTrue(finalResult.get() == 11);
+    }
+    
+    @Test
+    public void whenPassingPreviousStage_thenFunctionExecutionWithThenCompose() throws InterruptedException, ExecutionException {
+        CompletableFuture<Integer> finalResult = compute().thenCompose(this::computeAnother);
+        assertTrue(finalResult.get() == 20);
+    }
+    
+    public CompletableFuture<Integer> compute(){
+        return CompletableFuture.supplyAsync(() -> 10);
+    }
+    
+    public CompletableFuture<Integer> computeAnother(Integer i){
+        return CompletableFuture.supplyAsync(() -> 10 + i);
+    }
+
+}
@@ -0,0 +1,162 @@
+package com.baeldung.concurrent.executorservice;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class Java8ExecutorServiceIntegrationTest {
+
+    private Runnable runnableTask;
+    private Callable<String> callableTask;
+    private List<Callable<String>> callableTasks;
+
+    @Before
+    public void init() {
+
+        runnableTask = () -> {
+            try {
+                TimeUnit.MILLISECONDS.sleep(300);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        };
+
+        callableTask = () -> {
+            TimeUnit.MILLISECONDS.sleep(300);
+            return "Task's execution";
+        };
+
+        callableTasks = new ArrayList<>();
+        callableTasks.add(callableTask);
+        callableTasks.add(callableTask);
+        callableTasks.add(callableTask);
+    }
+
+    @Test
+    public void creationSubmittingTaskShuttingDown_whenShutDown_thenCorrect() {
+
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+        executorService.submit(runnableTask);
+        executorService.submit(callableTask);
+        executorService.shutdown();
+
+        assertTrue(executorService.isShutdown());
+    }
+
+    @Test
+    public void creationSubmittingTasksShuttingDownNow_whenShutDownAfterAwating_thenCorrect() {
+
+        ExecutorService threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
+
+        for (int i = 0; i < 100; i++) {
+            threadPoolExecutor.submit(callableTask);
+        }
+
+        List<Runnable> notExecutedTasks = smartShutdown(threadPoolExecutor);
+
+        assertTrue(threadPoolExecutor.isShutdown());
+        assertFalse(notExecutedTasks.isEmpty());
+        assertTrue(notExecutedTasks.size() < 98);
+    }
+
+    private List<Runnable> smartShutdown(ExecutorService executorService) {
+
+        List<Runnable> notExecutedTasks = new ArrayList<>();
+        executorService.shutdown();
+        try {
+            if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) {
+                notExecutedTasks = executorService.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            notExecutedTasks = executorService.shutdownNow();
+        }
+        return notExecutedTasks;
+    }
+
+    @Test
+    public void submittingTasks_whenExecutedOneAndAll_thenCorrect() {
+
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+        String result = null;
+        List<Future<String>> futures = new ArrayList<>();
+        try {
+            result = executorService.invokeAny(callableTasks);
+            futures = executorService.invokeAll(callableTasks);
+        } catch (InterruptedException | ExecutionException e) {
+            e.printStackTrace();
+        }
+
+        assertEquals("Task's execution", result);
+        assertTrue(futures.size() == 3);
+    }
+
+    @Test
+    public void submittingTaskShuttingDown_whenGetExpectedResult_thenCorrect() {
+
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+        Future<String> future = executorService.submit(callableTask);
+        String result = null;
+        try {
+            result = future.get();
+            result = future.get(200, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            e.printStackTrace();
+        }
+
+        executorService.shutdown();
+
+        assertEquals("Task's execution", result);
+    }
+
+    @Test
+    public void submittingTask_whenCanceled_thenCorrect() {
+
+        ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+        Future<String> future = executorService.submit(callableTask);
+
+        boolean canceled = future.cancel(true);
+        boolean isCancelled = future.isCancelled();
+
+        executorService.shutdown();
+
+        assertTrue(canceled);
+        assertTrue(isCancelled);
+    }
+
+    @Test
+    public void submittingTaskScheduling_whenExecuted_thenCorrect() {
+
+        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+
+        Future<String> resultFuture = executorService.schedule(callableTask, 1, TimeUnit.SECONDS);
+        String result = null;
+        try {
+            result = resultFuture.get();
+        } catch (InterruptedException | ExecutionException e) {
+            e.printStackTrace();
+        }
+
+        executorService.shutdown();
+
+        assertEquals("Task's execution", result);
+    }
+}
@@ -0,0 +1,145 @@
+package com.baeldung.concurrent.executorservice;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.*;
+
+import static junit.framework.TestCase.assertTrue;
+
+public class WaitingForThreadsToFinishManualTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WaitingForThreadsToFinishManualTest.class);
+    private final static ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(10);
+
+    public void awaitTerminationAfterShutdown(ExecutorService threadPool) {
+        threadPool.shutdown();
+        try {
+            if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
+                threadPool.shutdownNow();
+            }
+        } catch (InterruptedException ex) {
+            threadPool.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+    }
+    
+    @Test
+    public void givenMultipleThreads_whenUsingCountDownLatch_thenMainShoudWaitForAllToFinish() {
+
+        ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(10);
+        
+        try {
+            long startTime = System.currentTimeMillis();
+
+            // create a CountDownLatch that waits for the 2 threads to finish
+            CountDownLatch latch = new CountDownLatch(2);
+            
+            for (int i = 0; i < 2; i++) {
+                WORKER_THREAD_POOL.submit(() -> {
+                    try {
+                        Thread.sleep(1000);
+                        latch.countDown();
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                        Thread.currentThread().interrupt();
+                    }
+                });
+            }
+
+            // wait for the latch to be decremented by the two threads
+            latch.await();
+
+            long processingTime = System.currentTimeMillis() - startTime;
+            assertTrue(processingTime >= 1000);
+
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        awaitTerminationAfterShutdown(WORKER_THREAD_POOL);
+    }
+
+    @Test
+    public void givenMultipleThreads_whenInvokeAll_thenMainThreadShouldWaitForAllToFinish() {
+
+        ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(10);
+
+        List<Callable<String>> callables = Arrays.asList(
+            new DelayedCallable("fast thread", 100), 
+            new DelayedCallable("slow thread", 3000));
+
+        try {
+            long startProcessingTime = System.currentTimeMillis();
+            List<Future<String>> futures = WORKER_THREAD_POOL.invokeAll(callables);
+            
+            awaitTerminationAfterShutdown(WORKER_THREAD_POOL);
+
+            try {
+                WORKER_THREAD_POOL.submit((Callable<String>) () -> {
+                    Thread.sleep(1000000);
+                    return null;
+                });
+            } catch (RejectedExecutionException ex) {
+                //
+            }
+
+            long totalProcessingTime = System.currentTimeMillis() - startProcessingTime;
+            assertTrue(totalProcessingTime >= 3000);
+
+            String firstThreadResponse = futures.get(0)
+                .get();
+            assertTrue("First response should be from the fast thread", "fast thread".equals(firstThreadResponse));
+
+            String secondThreadResponse = futures.get(1)
+                .get();
+            assertTrue("Last response should be from the slow thread", "slow thread".equals(secondThreadResponse));
+
+        } catch (ExecutionException | InterruptedException ex) {
+            ex.printStackTrace();
+        }       
+    }
+
+    @Test
+    public void givenMultipleThreads_whenUsingCompletionService_thenMainThreadShouldWaitForAllToFinish() {
+
+        CompletionService<String> service = new ExecutorCompletionService<>(WORKER_THREAD_POOL);
+
+        List<Callable<String>> callables = Arrays.asList(
+            new DelayedCallable("fast thread", 100), 
+            new DelayedCallable("slow thread", 3000));
+
+        for (Callable<String> callable : callables) {
+            service.submit(callable);
+        }
+
+        try {
+
+            long startProcessingTime = System.currentTimeMillis();
+
+            Future<String> future = service.take();
+            String firstThreadResponse = future.get();
+            long totalProcessingTime = System.currentTimeMillis() - startProcessingTime;
+
+            assertTrue("First response should be from the fast thread", "fast thread".equals(firstThreadResponse));
+            assertTrue(totalProcessingTime >= 100 && totalProcessingTime < 1000);
+            LOG.debug("Thread finished after: " + totalProcessingTime + " milliseconds");
+
+            future = service.take();
+            String secondThreadResponse = future.get();
+            totalProcessingTime = System.currentTimeMillis() - startProcessingTime;
+
+            assertTrue("Last response should be from the slow thread", "slow thread".equals(secondThreadResponse));
+            assertTrue(totalProcessingTime >= 3000 && totalProcessingTime < 4000);
+            LOG.debug("Thread finished after: " + totalProcessingTime + " milliseconds");
+
+        } catch (ExecutionException | InterruptedException ex) {
+            ex.printStackTrace();
+        } finally {
+            awaitTerminationAfterShutdown(WORKER_THREAD_POOL);
+        }
+    }
+}
@@ -0,0 +1,53 @@
+package com.baeldung.concurrent.synchronize;
+
+import org.junit.Test;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.assertEquals;
+
+public class BaeldungSychronizedBlockUnitTest {
+
+    @Test
+    public void givenMultiThread_whenBlockSync() throws InterruptedException {
+        ExecutorService service = Executors.newFixedThreadPool(3);
+        BaeldungSynchronizedBlocks synchronizedBlocks = new BaeldungSynchronizedBlocks();
+
+        IntStream.range(0, 1000)
+          .forEach(count -> service.submit(synchronizedBlocks::performSynchronisedTask));
+        service.awaitTermination(500, TimeUnit.MILLISECONDS);
+
+        assertEquals(1000, synchronizedBlocks.getCount());
+    }
+
+    @Test
+    public void givenMultiThread_whenStaticSyncBlock() throws InterruptedException {
+        ExecutorService service = Executors.newCachedThreadPool();
+
+        IntStream.range(0, 1000)
+          .forEach(count -> service.submit(BaeldungSynchronizedBlocks::performStaticSyncTask));
+        service.awaitTermination(500, TimeUnit.MILLISECONDS);
+
+        assertEquals(1000, BaeldungSynchronizedBlocks.getStaticCount());
+    }
+
+    @Test
+    public void givenHoldingTheLock_whenReentrant_thenCanAcquireItAgain() {
+        Object lock = new Object();
+        synchronized (lock) {
+            System.out.println("First time acquiring it");
+
+            synchronized (lock) {
+                System.out.println("Entering again");
+
+                synchronized (lock) {
+                    System.out.println("And again");
+                }
+            }
+        }
+    }
+
+}
@@ -0,0 +1,51 @@
+package com.baeldung.concurrent.synchronize;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.assertEquals;
+
+public class BaeldungSynchronizeMethodsUnitTest {
+
+    @Test
+    @Ignore
+    public void givenMultiThread_whenNonSyncMethod() throws InterruptedException {
+        ExecutorService service = Executors.newFixedThreadPool(3);
+        BaeldungSynchronizedMethods method = new BaeldungSynchronizedMethods();
+
+        IntStream.range(0, 1000)
+          .forEach(count -> service.submit(method::calculate));
+        service.awaitTermination(100, TimeUnit.MILLISECONDS);
+
+        assertEquals(1000, method.getSum());
+    }
+
+    @Test
+    public void givenMultiThread_whenMethodSync() throws InterruptedException {
+        ExecutorService service = Executors.newFixedThreadPool(3);
+        BaeldungSynchronizedMethods method = new BaeldungSynchronizedMethods();
+
+        IntStream.range(0, 1000)
+          .forEach(count -> service.submit(method::synchronisedCalculate));
+        service.awaitTermination(100, TimeUnit.MILLISECONDS);
+
+        assertEquals(1000, method.getSyncSum());
+    }
+
+    @Test
+    public void givenMultiThread_whenStaticSyncMethod() throws InterruptedException {
+        ExecutorService service = Executors.newCachedThreadPool();
+
+        IntStream.range(0, 1000)
+          .forEach(count -> service.submit(BaeldungSynchronizedMethods::syncStaticCalculate));
+        service.awaitTermination(100, TimeUnit.MILLISECONDS);
+
+        assertEquals(1000, BaeldungSynchronizedMethods.staticSum);
+    }
+
+}
@@ -0,0 +1,55 @@
+package com.baeldung.concurrent.volatilekeyword;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class SharedObjectManualTest {
+
+    @Test
+    public void whenOneThreadWrites_thenVolatileReadsFromMainMemory() throws InterruptedException {
+        SharedObject sharedObject = new SharedObject();
+
+        Thread writer = new Thread(() -> sharedObject.incrementCount());
+        writer.start();
+        Thread.sleep(100);
+
+        Thread readerOne = new Thread(() -> {
+            int valueReadByThread2 = sharedObject.getCount();
+            assertEquals(1, valueReadByThread2);
+        });
+        readerOne.start();
+
+        Thread readerTwo = new Thread(() -> {
+            int valueReadByThread3 = sharedObject.getCount();
+            assertEquals(1, valueReadByThread3);
+        });
+        readerTwo.start();
+
+    }
+
+    @Test
+    public void whenTwoThreadWrites_thenVolatileReadsFromMainMemory() throws InterruptedException {
+        SharedObject sharedObject = new SharedObject();
+        Thread writerOne = new Thread(() -> sharedObject.incrementCount());
+        writerOne.start();
+        Thread.sleep(100);
+
+        Thread writerTwo = new Thread(() -> sharedObject.incrementCount());
+        writerTwo.start();
+        Thread.sleep(100);
+
+        Thread readerOne = new Thread(() -> {
+            int valueReadByThread2 = sharedObject.getCount();
+            assertEquals(2, valueReadByThread2);
+        });
+        readerOne.start();
+
+        Thread readerTwo = new Thread(() -> {
+            int valueReadByThread3 = sharedObject.getCount();
+            assertEquals(2, valueReadByThread3);
+        });
+        readerTwo.start();
+
+    }
+}
@@ -0,0 +1,66 @@
+package com.baeldung.concurrent.waitandnotify;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+import static org.junit.Assert.assertEquals;
+
+public class NetworkIntegrationTest {
+
+    private final ByteArrayOutputStream outContent = new ByteArrayOutputStream();
+    private final ByteArrayOutputStream errContent = new ByteArrayOutputStream();
+    private String expected;
+    
+    @Before
+    public void setUpStreams() {
+        System.setOut(new PrintStream(outContent));
+        System.setErr(new PrintStream(errContent));
+    }
+    
+    @Before
+    public void setUpExpectedOutput() {
+        StringWriter expectedStringWriter = new StringWriter();
+        
+        PrintWriter printWriter = new PrintWriter(expectedStringWriter);
+        printWriter.println("First packet");
+        printWriter.println("Second packet");
+        printWriter.println("Third packet");
+        printWriter.println("Fourth packet");
+        printWriter.close();
+        
+        expected = expectedStringWriter.toString();
+    }
+
+    @After
+    public void cleanUpStreams() {
+        System.setOut(null);
+        System.setErr(null);
+    }
+    
+    @Test
+    public void givenSenderAndReceiver_whenSendingPackets_thenNetworkSynchronized() {
+        Data data = new Data();
+        Thread sender = new Thread(new Sender(data));
+        Thread receiver = new Thread(new Receiver(data));
+        
+        sender.start();
+        receiver.start();
+        
+        //wait for sender and receiver to finish before we test against expected
+        try {
+            sender.join();
+            receiver.join();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt(); 
+            System.out.println("Thread Interrupted");
+        }
+        
+        assertEquals(expected, outContent.toString());
+    }
+}
@@ -0,0 +1,95 @@
+package com.baeldung.thread.join;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Demonstrates Thread.join behavior.
+ *
+ */
+public class ThreadJoinUnitTest {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadJoinUnitTest.class);
+
+    static class SampleThread extends Thread {
+        public int processingCount;
+
+        SampleThread(int processingCount) {
+            this.processingCount = processingCount;
+            LOGGER.debug("Thread " + this.getName() + " created");
+        }
+
+        @Override
+        public void run() {
+            LOGGER.debug("Thread " + this.getName() + " started");
+            while (processingCount > 0) {
+                try {
+                    Thread.sleep(1000); // Simulate some work being done by thread
+                } catch (InterruptedException e) {
+                    LOGGER.debug("Thread " + this.getName() + " interrupted.");
+                }
+                processingCount--;
+                LOGGER.debug("Inside Thread " + this.getName() + ", processingCount = " + processingCount);
+            }
+            LOGGER.debug("Thread " + this.getName() + " exiting");
+        }
+    }
+
+    @Test
+    public void givenNewThread_whenJoinCalled_returnsImmediately() throws InterruptedException {
+        Thread t1 = new SampleThread(0);
+        LOGGER.debug("Invoking join.");
+        t1.join();
+        LOGGER.debug("Returned from join");
+        LOGGER.debug("Thread state is" + t1.getState());
+        assertFalse(t1.isAlive());
+    }
+
+    @Test
+    public void givenStartedThread_whenJoinCalled_waitsTillCompletion()
+      throws InterruptedException {
+        Thread t2 = new SampleThread(1);
+        t2.start();
+        LOGGER.debug("Invoking join.");
+        t2.join();
+        LOGGER.debug("Returned from join");
+        assertFalse(t2.isAlive());
+    }
+
+    @Test
+    public void givenStartedThread_whenTimedJoinCalled_waitsUntilTimedout()
+      throws InterruptedException {
+        Thread t3 = new SampleThread(10);
+        t3.start();
+        t3.join(1000);
+        assertTrue(t3.isAlive());
+    }
+
+    @Test
+    @Ignore
+    public void givenThreadTerminated_checkForEffect_notGuaranteed()
+      throws InterruptedException {
+        SampleThread t4 = new SampleThread(10);
+        t4.start();
+      //not guaranteed to stop even if t4 finishes.
+        do {
+
+        } while (t4.processingCount > 0);
+    }
+
+    @Test
+    public void givenJoinWithTerminatedThread_checkForEffect_guaranteed()
+      throws InterruptedException {
+        SampleThread t4 = new SampleThread(10);
+        t4.start();
+        do {
+            t4.join(100);
+        } while (t4.processingCount > 0);
+    }
+
+}