Merge branch 'master' of https://github.com/eugenp/tutorials into task/BAEL-9709

2018-10-26 23:56:29 +05:30
parent 8c361330f1 362bd203e9
commit bbc4cdd3f7
47 changed files with 1329 additions and 94 deletions
@@ -7,7 +7,7 @@ public class BubbleSort {
    void bubbleSort(Integer[] arr) {
        int n = arr.length;
        IntStream.range(0, n - 1)
-          .flatMap(i -> IntStream.range(i + 1, n - i))
+          .flatMap(i -> IntStream.range(1, n - i))
          .forEach(j -> {
              if (arr[j - 1] > arr[j]) {
                  int temp = arr[j];
@@ -1,8 +1,8 @@
 package com.baeldung.algorithms.bubblesort;

-import static org.junit.Assert.*;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;

-import org.junit.Test;
+import org.junit.jupiter.api.Test;

 public class BubbleSortUnitTest {

@@ -1,13 +1,20 @@
 package com.baeldung.java8;

+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.function.Consumer;
+
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import java.util.ArrayList;
-import java.util.List;
-import java.util.function.Consumer;
-
 public class Java8ForEachUnitTest {

    private static final Logger LOG = LoggerFactory.getLogger(Java8ForEachUnitTest.class);
@@ -29,8 +36,18 @@ public class Java8ForEachUnitTest {
        }

        // Java 8 - forEach
-        LOG.debug("--- forEach method ---");
-        names.forEach(name -> LOG.debug(name));
+        names.forEach(name -> {
+            System.out.println(name);
+        });
+
+        LOG.debug("--- Print Consumer ---");
+        Consumer<String> printConsumer = new Consumer<String>() {
+            public void accept(String name) {
+                System.out.println(name);
+            };
+        };
+
+        names.forEach(printConsumer);

        // Anonymous inner class that implements Consumer interface
        LOG.debug("--- Anonymous inner class ---");
@@ -40,17 +57,55 @@ public class Java8ForEachUnitTest {
            }
        });

-        // Create a Consumer implementation to then use in a forEach method
-        Consumer<String> consumerNames = name -> {
-            LOG.debug(name);
-        };
-        LOG.debug("--- Implementation of Consumer interface ---");
-        names.forEach(consumerNames);
+        // Java 8 - forEach - Lambda Syntax
+        LOG.debug("--- forEach method ---");
+        names.forEach(name -> LOG.debug(name));

-        // Print elements using a Method Reference
+        // Java 8 - forEach - Print elements using a Method Reference
        LOG.debug("--- Method Reference ---");
        names.forEach(LOG::debug);
+    }

+    @Test
+    public void givenList_thenIterateAndPrintResults() {
+        List<String> names = Arrays.asList("Larry", "Steve", "James");
+
+        names.forEach(System.out::println);
+    }
+
+    @Test
+    public void givenSet_thenIterateAndPrintResults() {
+        Set<String> uniqueNames = new HashSet<>(Arrays.asList("Larry", "Steve", "James"));
+
+        uniqueNames.forEach(System.out::println);
+    }
+
+    @Test
+    public void givenQueue_thenIterateAndPrintResults() {
+        Queue<String> namesQueue = new ArrayDeque<>(Arrays.asList("Larry", "Steve", "James"));
+
+        namesQueue.forEach(System.out::println);
+    }
+
+    @Test
+    public void givenMap_thenIterateAndPrintResults() {
+        Map<Integer, String> namesMap = new HashMap<>();
+        namesMap.put(1, "Larry");
+        namesMap.put(2, "Steve");
+        namesMap.put(3, "James");
+
+        namesMap.entrySet()
+            .forEach(entry -> System.out.println(entry.getKey() + " " + entry.getValue()));
+    }
+
+    @Test
+    public void givenMap_whenUsingBiConsumer_thenIterateAndPrintResults2() {
+        Map<Integer, String> namesMap = new HashMap<>();
+        namesMap.put(1, "Larry");
+        namesMap.put(2, "Steve");
+        namesMap.put(3, "James");
+
+        namesMap.forEach((key, value) -> System.out.println(key + " " + value));
    }

 }
@@ -154,6 +154,12 @@
            <artifactId>async-http-client</artifactId>
            <version>${async-http-client.version}</version>
        </dependency>
+        <dependency>
+            <groupId>com.opencsv</groupId>
+            <artifactId>opencsv</artifactId>
+            <version>${opencsv.version}</version>
+            <scope>test</scope>
+        </dependency>
    </dependencies>

    <build>
@@ -247,7 +253,7 @@
        <protonpack.version>1.13</protonpack.version>
        <streamex.version>0.6.5</streamex.version>
        <vavr.version>0.9.0</vavr.version>
-
+        <opencsv.version>4.1</opencsv.version>
        <!-- testing -->
        <assertj.version>3.6.1</assertj.version>
        <avaitility.version>1.7.0</avaitility.version>
@@ -9,13 +9,13 @@ import java.util.zip.ZipInputStream;

 public class UnzipFile {
    public static void main(final String[] args) throws IOException {
-        final String fileZip = "src/main/resources/compressed.zip";
+        final String fileZip = "src/main/resources/unzipTest/compressed.zip";
+        final File destDir = new File("src/main/resources/unzipTest");
        final byte[] buffer = new byte[1024];
        final ZipInputStream zis = new ZipInputStream(new FileInputStream(fileZip));
        ZipEntry zipEntry = zis.getNextEntry();
        while (zipEntry != null) {
-            final String fileName = zipEntry.getName();
-            final File newFile = new File("src/main/resources/unzipTest/" + fileName);
+            final File newFile = newFile(destDir, zipEntry);
            final FileOutputStream fos = new FileOutputStream(newFile);
            int len;
            while ((len = zis.read(buffer)) > 0) {
@@ -27,4 +27,20 @@ public class UnzipFile {
        zis.closeEntry();
        zis.close();
    }
+    
+    /**
+     * @see https://snyk.io/research/zip-slip-vulnerability
+     */
+    public static File newFile(File destinationDir, ZipEntry zipEntry) throws IOException {
+        File destFile = new File(destinationDir, zipEntry.getName());
+        
+        String destDirPath = destinationDir.getCanonicalPath();
+        String destFilePath = destFile.getCanonicalPath();
+        
+        if (!destFilePath.startsWith(destDirPath + File.separator)) {
+            throw new IOException("Entry is outside of the target dir: " + zipEntry.getName());
+        }
+        
+        return destFile;
+    }
 }
@@ -0,0 +1,106 @@
+package com.baeldung.csv;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Scanner;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.opencsv.CSVReader;
+
+public class ReadCSVInArrayUnitTest {
+    public static final String COMMA_DELIMITER = ",";
+    public static final String CSV_FILE = "src/test/resources/book.csv";
+    public static final List<List<String>> EXPECTED_ARRAY = Collections.unmodifiableList(new ArrayList<List<String>>() {
+        {
+            add(new ArrayList<String>() {
+                {
+                    add("Mary Kom");
+                    add("Unbreakable");
+                }
+            });
+            add(new ArrayList<String>() {
+                {
+                    add("Kapil Isapuari");
+                    add("Farishta");
+                }
+            });
+        }
+    });
+
+    @Test
+    public void givenCSVFile_whenBufferedReader_thenContentsAsExpected() throws IOException {
+        List<List<String>> records = new ArrayList<List<String>>();
+        try (BufferedReader br = new BufferedReader(new FileReader(CSV_FILE))) {
+            String line = "";
+            while ((line = br.readLine()) != null) {
+                String[] values = line.split(COMMA_DELIMITER);
+                records.add(Arrays.asList(values));
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        for (int i = 0; i < EXPECTED_ARRAY.size(); i++) {
+            Assert.assertArrayEquals(EXPECTED_ARRAY.get(i)
+                .toArray(),
+                records.get(i)
+                    .toArray());
+        }
+    }
+
+    @Test
+    public void givenCSVFile_whenScanner_thenContentsAsExpected() throws IOException {
+        List<List<String>> records = new ArrayList<List<String>>();
+        try (Scanner scanner = new Scanner(new File(CSV_FILE));) {
+            while (scanner.hasNextLine()) {
+                records.add(getRecordFromLine(scanner.nextLine()));
+            }
+        } catch (FileNotFoundException e) {
+            e.printStackTrace();
+        }
+        for (int i = 0; i < EXPECTED_ARRAY.size(); i++) {
+            Assert.assertArrayEquals(EXPECTED_ARRAY.get(i)
+                .toArray(),
+                records.get(i)
+                    .toArray());
+        }
+    }
+
+    private List<String> getRecordFromLine(String line) {
+        List<String> values = new ArrayList<String>();
+        try (Scanner rowScanner = new Scanner(line)) {
+            rowScanner.useDelimiter(COMMA_DELIMITER);
+            while (rowScanner.hasNext()) {
+                values.add(rowScanner.next());
+            }
+        }
+        return values;
+    }
+
+    @Test
+    public void givenCSVFile_whenOpencsv_thenContentsAsExpected() throws IOException {
+        List<List<String>> records = new ArrayList<List<String>>();
+        try (CSVReader csvReader = new CSVReader(new FileReader(CSV_FILE));) {
+            String[] values = null;
+            while ((values = csvReader.readNext()) != null) {
+                records.add(Arrays.asList(values));
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        for (int i = 0; i < EXPECTED_ARRAY.size(); i++) {
+            Assert.assertArrayEquals(EXPECTED_ARRAY.get(i)
+                .toArray(),
+                records.get(i)
+                    .toArray());
+        }
+    }
+}
@@ -0,0 +1,2 @@
+Mary Kom,Unbreakable
+Kapil Isapuari,Farishta
@@ -0,0 +1,41 @@
+package com.baeldung.string;
+
+import java.math.RoundingMode;
+import java.text.DecimalFormat;
+import java.text.NumberFormat;
+
+public class DoubleToString {
+
+    public static String truncateByCast(double d) {
+        return String.valueOf((int) d);
+    }
+
+    public static String roundWithStringFormat(double d) {
+        return String.format("%.0f", d);
+    }
+
+    public static String truncateWithNumberFormat(double d) {
+        NumberFormat nf = NumberFormat.getInstance();
+        nf.setMaximumFractionDigits(0);
+        nf.setRoundingMode(RoundingMode.FLOOR);
+        return nf.format(d);
+    }
+
+    public static String roundWithNumberFormat(double d) {
+        NumberFormat nf = NumberFormat.getInstance();
+        nf.setMaximumFractionDigits(0);
+        return nf.format(d);
+    }
+
+    public static String truncateWithDecimalFormat(double d) {
+        DecimalFormat df = new DecimalFormat("#,###");
+        df.setRoundingMode(RoundingMode.FLOOR);
+        return df.format(d);
+    }
+
+    public static String roundWithDecimalFormat(double d) {
+        DecimalFormat df = new DecimalFormat("#,###");
+        return df.format(d);
+    }
+
+}
@@ -1,4 +1,4 @@
-package com.baeldung;
+package com.baeldung.simpledateformat;

 import org.junit.Test;

@@ -32,6 +32,7 @@ public class SimpleDateFormatUnitTest {
  @Test
  public void givenStringDate_whenParsed_thenCheckDateCorrect() throws Exception{
    SimpleDateFormat formatter = new SimpleDateFormat("dd-MM-yyyy");
+    formatter.setTimeZone(TimeZone.getTimeZone("Europe/London"));
    Date myDate = new Date(233276400000L);
    Date parsedDate = formatter.parse("24-05-1977");
    assertEquals(myDate.getTime(), parsedDate.getTime());
@@ -0,0 +1,45 @@
+package com.baeldung.string;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+
+public class DoubleToStringUnitTest {
+    
+    private static final double DOUBLE_VALUE = 3.56;
+    private static final String TRUNCATED_DOUBLE = "3";
+    private static final String ROUNDED_UP_DOUBLE = "4";
+    
+    
+    @Test
+    public void truncateByCastTest() {
+        assertThat(DoubleToString.truncateByCast(DOUBLE_VALUE)).isEqualTo(TRUNCATED_DOUBLE);
+    }
+    
+    @Test
+    public void roundingWithStringFormatTest() {
+        assertThat(DoubleToString.roundWithStringFormat(DOUBLE_VALUE)).isEqualTo(ROUNDED_UP_DOUBLE);
+    }
+
+    @Test
+    public void truncateWithNumberFormatTest() {
+        assertThat(DoubleToString.truncateWithNumberFormat(DOUBLE_VALUE)).isEqualTo(TRUNCATED_DOUBLE);
+    }
+    
+    @Test
+    public void roundWithNumberFormatTest() {
+        assertThat(DoubleToString.roundWithNumberFormat(DOUBLE_VALUE)).isEqualTo(ROUNDED_UP_DOUBLE);
+    }
+
+    @Test
+    public void truncateWithDecimalFormatTest() {
+        assertThat(DoubleToString.truncateWithDecimalFormat(DOUBLE_VALUE)).isEqualTo(TRUNCATED_DOUBLE);
+    }
+    
+    @Test
+    public void roundWithDecimalFormatTest() {
+        assertThat(DoubleToString.roundWithDecimalFormat(DOUBLE_VALUE)).isEqualTo(ROUNDED_UP_DOUBLE);
+    }
+
+
+}
@@ -0,0 +1,8 @@
+package com.baeldung.thread
+
+class SimpleRunnable: Runnable {
+
+    override fun run() {
+        println("${Thread.currentThread()} has run.")
+    }
+}
@@ -0,0 +1,8 @@
+package com.baeldung.thread
+
+class SimpleThread: Thread() {
+
+    override fun run() {
+        println("${Thread.currentThread()} has run.")
+    }
+}
@@ -24,14 +24,15 @@ internal class SliceTest {
        assertIterableEquals(expected, actual)
    }

-    @Test
-    fun whenSlicingBeyondTheRangeOfTheArray_thenContainManyNulls() {
-        val original = arrayOf(12, 3, 34, 4)
-        val actual = original.slice(3..8)
-        val expected = listOf(4, null, null, null, null, null)
-
-        assertIterableEquals(expected, actual)
-    }
+// From the 1.3 version of Kotlin APIs, slice doesn't return array of nulls but throw IndexOutOfBoundsException
+//    @Test
+//    fun whenSlicingBeyondTheRangeOfTheArray_thenContainManyNulls() {
+//        val original = arrayOf(12, 3, 34, 4)
+//        val actual = original.slice(3..8)
+//        val expected = listOf(4, null, null, null, null, null)
+//
+//        assertIterableEquals(expected, actual)
+//    }

    @Test
    fun whenSlicingBeyondRangeOfArrayWithStep_thenOutOfBoundsException() {
@@ -1,16 +1,12 @@
 package com.baeldung.fuel

-import awaitObjectResult
-import awaitStringResponse
 import com.github.kittinunf.fuel.Fuel
 import com.github.kittinunf.fuel.core.FuelManager
-import com.github.kittinunf.fuel.core.Request
 import com.github.kittinunf.fuel.core.interceptors.cUrlLoggingRequestInterceptor
 import com.github.kittinunf.fuel.gson.responseObject
 import com.github.kittinunf.fuel.httpGet
 import com.github.kittinunf.fuel.rx.rx_object
 import com.google.gson.Gson
-import kotlinx.coroutines.experimental.runBlocking
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Test
 import java.io.File
@@ -226,32 +222,26 @@ internal class FuelHttpUnitTest {

    }

-    @Test
-    fun whenMakeGETRequestUsingCoroutines_thenResponseStatusCode200() {

-        runBlocking {
-            val (request, response, result) = Fuel.get("http://httpbin.org/get").awaitStringResponse()
+// The new 1.3 coroutine APIs, aren't implemented yet in Fuel Library
+//    @Test
+//    fun whenMakeGETRequestUsingCoroutines_thenResponseStatusCode200() = runBlocking {
+//            val (request, response, result) = Fuel.get("http://httpbin.org/get").awaitStringResponse()
+//
+//            result.fold({ data ->
+//                Assertions.assertEquals(200, response.statusCode)
+//
+//            }, { error -> })
+//    }

-            result.fold({ data ->
-                Assertions.assertEquals(200, response.statusCode)
-
-            }, { error -> })
-        }
-
-    }
-
-    @Test
-    fun whenMakeGETRequestUsingCoroutines_thenDeserializeResponse() {
-
-
-        runBlocking {
-            Fuel.get("https://jsonplaceholder.typicode.com/posts?id=1").awaitObjectResult(Post.Deserializer())
-                    .fold({ data ->
-                        Assertions.assertEquals(1, data.get(0).userId)
-                    }, { error -> })
-        }
-
-    }
+// The new 1.3 coroutine APIs, aren't implemented yet in Fuel Library
+//    @Test
+//    fun whenMakeGETRequestUsingCoroutines_thenDeserializeResponse() = runBlocking {
+//            Fuel.get("https://jsonplaceholder.typicode.com/posts?id=1").awaitObjectResult(Post.Deserializer())
+//                    .fold({ data ->
+//                        Assertions.assertEquals(1, data.get(0).userId)
+//                    }, { error -> })
+//        }

    @Test
    fun whenMakeGETPostRequestUsingRoutingAPI_thenDeserializeResponse() {
@@ -1,9 +1,8 @@
 package com.baeldung.kotlin

-import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.*
 import org.junit.Test
 import java.util.concurrent.atomic.AtomicInteger
-import kotlin.coroutines.experimental.buildSequence
 import kotlin.system.measureTimeMillis
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
@@ -14,7 +13,7 @@ class CoroutinesTest {
    @Test
    fun givenBuildSequence_whenTakeNElements_thenShouldReturnItInALazyWay() {
        //given
-        val fibonacciSeq = buildSequence {
+        val fibonacciSeq = sequence {
            var a = 0
            var b = 1

@@ -39,7 +38,7 @@ class CoroutinesTest {
    @Test
    fun givenLazySeq_whenTakeNElements_thenShouldReturnAllElements() {
        //given
-        val lazySeq = buildSequence {
+        val lazySeq = sequence {
            print("START ")
            for (i in 1..5) {
                yield(i)
@@ -60,8 +59,8 @@ class CoroutinesTest {
        val res = mutableListOf<String>()

        //when
-        runBlocking<Unit> {
-            val promise = launch(CommonPool) { expensiveComputation(res) }
+        runBlocking {
+            val promise = launch(Dispatchers.Default) { expensiveComputation(res) }
            res.add("Hello,")
            promise.join()
        }
@@ -85,7 +84,7 @@ class CoroutinesTest {

            //when
            val jobs = List(numberOfCoroutines) {
-                launch(CommonPool) {
+                launch(Dispatchers.Default) {
                    delay(1L)
                    counter.incrementAndGet()
                }
@@ -101,7 +100,7 @@ class CoroutinesTest {
    fun givenCancellableJob_whenRequestForCancel_thenShouldQuit() {
        runBlocking<Unit> {
            //given
-            val job = launch(CommonPool) {
+            val job = launch(Dispatchers.Default) {
                while (isActive) {
                    //println("is working")
                }
@@ -135,8 +134,8 @@ class CoroutinesTest {
            val delay = 1000L
            val time = measureTimeMillis {
                //given
-                val one = async(CommonPool) { someExpensiveComputation(delay) }
-                val two = async(CommonPool) { someExpensiveComputation(delay) }
+                val one = async(Dispatchers.Default) { someExpensiveComputation(delay) }
+                val two = async(Dispatchers.Default) { someExpensiveComputation(delay) }

                //when
                runBlocking {
@@ -156,8 +155,8 @@ class CoroutinesTest {
            val delay = 1000L
            val time = measureTimeMillis {
                //given
-                val one = async(CommonPool, CoroutineStart.LAZY) { someExpensiveComputation(delay) }
-                val two = async(CommonPool, CoroutineStart.LAZY) { someExpensiveComputation(delay) }
+                val one = async(Dispatchers.Default, CoroutineStart.LAZY) { someExpensiveComputation(delay) }
+                val two = async(Dispatchers.Default, CoroutineStart.LAZY) { someExpensiveComputation(delay) }

                //when
                runBlocking {
@@ -0,0 +1,52 @@
+package com.baeldung.thread
+
+import kotlinx.coroutines.*
+import org.junit.jupiter.api.Test
+
+class CoroutineUnitTest {
+
+    @Test
+    fun whenCreateCoroutineWithLaunchWithoutContext_thenRun() = runBlocking {
+
+        val job = launch {
+            println("${Thread.currentThread()} has run.")
+        }
+
+    }
+
+    @Test
+    fun whenCreateCoroutineWithLaunchWithDefaultContext_thenRun() = runBlocking {
+
+        val job = launch(Dispatchers.Default) {
+            println("${Thread.currentThread()} has run.")
+        }
+    }
+
+    @Test
+    fun whenCreateCoroutineWithLaunchWithUnconfinedContext_thenRun() = runBlocking {
+
+        val job = launch(Dispatchers.Unconfined) {
+            println("${Thread.currentThread()} has run.")
+        }
+    }
+
+    @Test
+    fun whenCreateCoroutineWithLaunchWithDedicatedThread_thenRun() = runBlocking {
+
+        val job = launch(newSingleThreadContext("dedicatedThread")) {
+            println("${Thread.currentThread()} has run.")
+        }
+
+    }
+
+    @Test
+    fun whenCreateAsyncCoroutine_thenRun() = runBlocking {
+
+        val deferred = async(Dispatchers.IO) {
+            return@async "${Thread.currentThread()} has run."
+        }
+
+        val result = deferred.await()
+        println(result)
+    }
+}
@@ -0,0 +1,38 @@
+package com.baeldung.thread
+
+import org.junit.jupiter.api.Test
+import kotlin.concurrent.thread
+
+class ThreadUnitTest {
+
+    @Test
+    fun whenCreateThread_thenRun() {
+
+        val thread = SimpleThread()
+        thread.start()
+    }
+
+    @Test
+    fun whenCreateThreadWithRunnable_thenRun() {
+
+        val threadWithRunnable = Thread(SimpleRunnable())
+        threadWithRunnable.start()
+    }
+
+    @Test
+    fun whenCreateThreadWithSAMConversions_thenRun() {
+
+        val thread = Thread {
+            println("${Thread.currentThread()} has run.")
+        }
+        thread.start()
+    }
+
+    @Test
+    fun whenCreateThreadWithMethodExtension_thenRun() {
+
+        thread(start = true) {
+            println("${Thread.currentThread()} has run.")
+        }
+    }
+}
@@ -11,7 +11,7 @@ import java.time.LocalDate;

 import static org.hibernate.testing.transaction.TransactionUtil.doInHibernate;

-public class HibernateCustomTypesUnitTest {
+public class HibernateCustomTypesManualTest {

    @Test
    public void givenEmployee_whenSavedWithCustomTypes_thenEntityIsSaved() throws IOException {
@@ -1,19 +1,22 @@
 package com.baeldung.zoneddatetime;

+import static org.junit.jupiter.api.Assertions.assertThrows;
+
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 import java.util.logging.Logger;

-import org.junit.Test;
+import org.junit.jupiter.api.Test;

 public class ZonedDateTimeUnitTest {

    private static final Logger log = Logger.getLogger(ZonedDateTimeUnitTest.class.getName());

    @Test
-    public void testZonedDateTimeToString() {
+    public void givenZonedDateTime_whenConvertToString_thenOk() {

        ZonedDateTime zonedDateTimeNow = ZonedDateTime.now(ZoneId.of("UTC"));
        ZonedDateTime zonedDateTimeOf = ZonedDateTime.of(2018, 01, 01, 0, 0, 0, 0, ZoneId.of("UTC"));
@@ -21,10 +24,10 @@ public class ZonedDateTimeUnitTest {
        LocalDateTime localDateTime = LocalDateTime.now();
        ZonedDateTime zonedDateTime = ZonedDateTime.of(localDateTime, ZoneId.of("UTC"));

-        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy - hh:mm:ss Z");
+        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy - HH:mm:ss Z");
        String formattedString = zonedDateTime.format(formatter);

-        DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("MM/dd/yyyy - hh:mm:ss z");
+        DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("MM/dd/yyyy - HH:mm:ss z");
        String formattedString2 = zonedDateTime.format(formatter2);

        log.info(formattedString);
@@ -33,9 +36,21 @@ public class ZonedDateTimeUnitTest {
    }

    @Test
-    public void testZonedDateTimeFromString() {
+    public void givenString_whenParseZonedDateTime_thenOk() {
+        ZonedDateTime zonedDateTime = ZonedDateTime.parse("2011-12-03T10:15:30+01:00");

-        ZonedDateTime zonedDateTime = ZonedDateTime.parse("2011-12-03T10:15:30+01:00", DateTimeFormatter.ISO_ZONED_DATE_TIME);
+        log.info(zonedDateTime.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
+    }
+    
+    @Test
+    public void givenString_whenParseZonedDateTimeWithoutZone_thenException() {
+        assertThrows(DateTimeParseException.class, () -> ZonedDateTime.parse("2011-12-03T10:15:30", DateTimeFormatter.ISO_DATE_TIME));
+    }
+    
+    @Test
+    public void givenString_whenParseLocalDateTimeAtZone_thenOk() {
+        ZoneId timeZone = ZoneId.systemDefault();
+        ZonedDateTime zonedDateTime = LocalDateTime.parse("2011-12-03T10:15:30", DateTimeFormatter.ISO_DATE_TIME).atZone(timeZone);

        log.info(zonedDateTime.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
    }
@@ -259,6 +259,11 @@
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
+        <dependency>
+            <groupId>org.apache.storm</groupId>
+            <artifactId>storm-core</artifactId>
+            <version>${storm.version}</version>
+        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
@@ -432,6 +437,7 @@
    </repositories>

    <properties>
+        <storm.version>1.2.2</storm.version>
        <kryo.version>4.0.1</kryo.version>
        <h2.version>1.4.196</h2.version>
        <reladomo.version>16.5.1</reladomo.version>
@@ -0,0 +1,34 @@
+package com.baeldung.storm;
+
+import com.baeldung.storm.bolt.AggregatingBolt;
+import com.baeldung.storm.bolt.FileWritingBolt;
+import com.baeldung.storm.bolt.FilteringBolt;
+import com.baeldung.storm.bolt.PrintingBolt;
+import com.baeldung.storm.spout.RandomNumberSpout;
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+
+public class TopologyRunner {
+    public static void main(String[] args) {
+       runTopology();
+    }
+
+    public static void runTopology() {
+        String filePath = "./src/main/resources/operations.txt";
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("randomNumberSpout", new RandomNumberSpout());
+        builder.setBolt("filteringBolt", new FilteringBolt()).shuffleGrouping("randomNumberSpout");
+        builder.setBolt("aggregatingBolt", new AggregatingBolt()
+          .withTimestampField("timestamp")
+          .withLag(BaseWindowedBolt.Duration.seconds(1))
+          .withWindow(BaseWindowedBolt.Duration.seconds(5))).shuffleGrouping("filteringBolt");
+        builder.setBolt("fileBolt", new FileWritingBolt(filePath)).shuffleGrouping("aggregatingBolt");
+
+        Config config = new Config();
+        config.setDebug(false);
+        LocalCluster cluster = new LocalCluster();
+        cluster.submitTopology("Test", config, builder.createTopology());
+    }
+}
@@ -0,0 +1,39 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.windowing.TupleWindow;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+
+public class AggregatingBolt extends BaseWindowedBolt {
+    private OutputCollector outputCollector;
+    @Override
+    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+        this.outputCollector = collector;
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(new Fields("sumOfOperations", "beginningTimestamp", "endTimestamp"));
+    }
+
+    @Override
+    public void execute(TupleWindow tupleWindow) {
+        List<Tuple> tuples = tupleWindow.get();
+        tuples.sort(Comparator.comparing(a -> a.getLongByField("timestamp")));
+        //This is safe since the window is calculated basing on Tuple's timestamp, thus it can't really be empty
+        Long beginningTimestamp = tuples.get(0).getLongByField("timestamp");
+        Long endTimestamp = tuples.get(tuples.size() - 1).getLongByField("timestamp");
+        int sumOfOperations = tuples.stream().mapToInt(tuple -> tuple.getIntegerByField("operation")).sum();
+        Values values = new Values(sumOfOperations, beginningTimestamp, endTimestamp);
+        outputCollector.emit(values);
+    }
+}
@@ -0,0 +1,72 @@
+package com.baeldung.storm.bolt;
+
+import com.baeldung.storm.model.AggregatedWindow;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+
+public class FileWritingBolt extends BaseRichBolt {
+    public static Logger logger = LoggerFactory.getLogger(FileWritingBolt.class);
+    private BufferedWriter writer;
+    private String filePath;
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
+        objectMapper = new ObjectMapper();
+        objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
+        try {
+            writer = new BufferedWriter(new FileWriter(filePath));
+        } catch (IOException e) {
+            logger.error("Failed to open a file for writing.", e);
+        }
+    }
+
+    @Override
+    public void execute(Tuple tuple) {
+        int sumOfOperations = tuple.getIntegerByField("sumOfOperations");
+        long beginningTimestamp = tuple.getLongByField("beginningTimestamp");
+        long endTimestamp = tuple.getLongByField("endTimestamp");
+
+        if(sumOfOperations > 200) {
+            AggregatedWindow aggregatedWindow = new AggregatedWindow(sumOfOperations, beginningTimestamp, endTimestamp);
+            try {
+                writer.write(objectMapper.writeValueAsString(aggregatedWindow));
+                writer.write("\n");
+                writer.flush();
+            } catch (IOException e) {
+                logger.error("Failed to write data to file.", e);
+            }
+        }
+    }
+
+    public FileWritingBolt(String filePath) {
+        this.filePath = filePath;
+    }
+
+    @Override
+    public void cleanup() {
+        try {
+            writer.close();
+        } catch (IOException e) {
+            logger.error("Failed to close the writer!");
+        }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+
+    }
+}
@@ -0,0 +1,22 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+
+public class FilteringBolt extends BaseBasicBolt {
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+        int operation = tuple.getIntegerByField("operation");
+        if(operation > 0 ) {
+            basicOutputCollector.emit(tuple.getValues());
+        }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("operation", "timestamp"));
+    }
+}
@@ -0,0 +1,18 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Tuple;
+
+public class PrintingBolt extends BaseBasicBolt {
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+        System.out.println(tuple);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+
+    }
+}
@@ -0,0 +1,16 @@
+package com.baeldung.storm.model;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+@JsonSerialize
+public class AggregatedWindow {
+    int sumOfOperations;
+    long beginningTimestamp;
+    long endTimestamp;
+
+    public AggregatedWindow(int sumOfOperations, long beginningTimestamp, long endTimestamp) {
+        this.sumOfOperations = sumOfOperations;
+        this.beginningTimestamp = beginningTimestamp;
+        this.endTimestamp = endTimestamp;
+    }
+}
@@ -0,0 +1,40 @@
+package com.baeldung.storm.model;
+
+public class User {
+    private String username;
+    private String password;
+    private String email;
+    private int age;
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getEmail() {
+        return email;
+    }
+
+    public void setEmail(String email) {
+        this.email = email;
+    }
+
+    public int getAge() {
+        return age;
+    }
+
+    public void setAge(int age) {
+        this.age = age;
+    }
+}
@@ -0,0 +1,30 @@
+package com.baeldung.storm.serialization;
+
+
+import com.baeldung.storm.model.User;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class UserSerializer extends Serializer<User>{
+    @Override
+    public void write(Kryo kryo, Output output, User user) {
+        output.writeString(user.getEmail());
+        output.writeString(user.getUsername());
+        output.write(user.getAge());
+    }
+
+    @Override
+    public User read(Kryo kryo, Input input, Class<User> aClass) {
+        User user = new User();
+        String email = input.readString();
+        String name = input.readString();
+        int age = input.read();
+        user.setAge(age);
+        user.setEmail(email);
+        user.setUsername(name);
+
+        return  user;
+    }
+}
@@ -0,0 +1,35 @@
+package com.baeldung.storm.spout;
+
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.Map;
+import java.util.Random;
+
+public class RandomIntSpout extends BaseRichSpout {
+
+    private Random random;
+    private SpoutOutputCollector outputCollector;
+
+    @Override
+    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
+        random = new Random();
+        outputCollector = spoutOutputCollector;
+    }
+
+    @Override
+    public void nextTuple() {
+        Utils.sleep(1000);
+        outputCollector.emit(new Values(random.nextInt(), System.currentTimeMillis()));
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("randomInt", "timestamp"));
+    }
+}
--- a/Show More
+++ b/Show More