删除编译错误

2024-04-30 12:20:08 -04:00
parent f857c68a7c
commit 83883319ab
243 changed files with 16741 additions and 0 deletions
@@ -0,0 +1,14 @@
+## Ratpack
+
+This module contains articles about Ratpack.
+
+### Relevant articles
+
+- [Introduction to Ratpack](https://www.baeldung.com/ratpack)
+- [Ratpack Google Guice Integration](https://www.baeldung.com/ratpack-google-guice)
+- [Ratpack Integration with Spring Boot](http://www.baeldung.com/ratpack-spring-boot)
+- [Ratpack with Hystrix](https://www.baeldung.com/ratpack-hystrix)
+- [Ratpack HTTP Client](https://www.baeldung.com/ratpack-http-client)
+- [Ratpack with RxJava](https://www.baeldung.com/ratpack-rxjava)
+- [Ratpack with Groovy](https://www.baeldung.com/ratpack-groovy)
+- [Reactive Streams API with Ratpack](https://www.baeldung.com/ratpack-reactive-streams-api)
@@ -0,0 +1,37 @@
+buildscript {
+  repositories {
+    jcenter()
+  }
+  dependencies {
+    classpath "io.ratpack:ratpack-gradle:1.5.4"
+    classpath "com.h2database:h2:1.4.193"
+  }
+}
+
+if (!JavaVersion.current().java8Compatible) {
+  throw new IllegalStateException("Must be built with Java 8 or higher")
+}
+
+apply plugin: "io.ratpack.ratpack-java"
+apply plugin: 'java'
+apply plugin: 'groovy'
+apply plugin: 'io.ratpack.ratpack-groovy'
+
+repositories {
+  jcenter()
+}
+
+dependencies {
+  compile ratpack.dependency('hikari')
+  compile 'com.h2database:h2:1.4.193'
+  testCompile 'junit:junit:4.11'
+  runtime "org.slf4j:slf4j-simple:1.7.21"
+}
+
+test {
+    testLogging {
+        events 'started', 'passed'
+    }
+}
+
+mainClassName = "com.baeldung.Application"
@@ -0,0 +1,111 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>ratpack</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <name>ratpack</name>
+    <packaging>jar</packaging>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>web-modules</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-spring-boot-starter</artifactId>
+            <version>${ratpack.version}</version>
+            <type>pom</type>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-core</artifactId>
+            <version>${ratpack.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.codehaus.groovy</groupId>
+            <artifactId>groovy-sql</artifactId>
+            <version>${groovy.sql.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-hikari</artifactId>
+            <version>${ratpack.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-groovy-test</artifactId>
+            <version>${ratpack.test.latest.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-hystrix</artifactId>
+            <version>${ratpack.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.netflix.hystrix</groupId>
+                    <artifactId>hystrix-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.netflix.hystrix</groupId>
+            <artifactId>hystrix-core</artifactId>
+            <version>${hystrix.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-rx</artifactId>
+            <version>${ratpack.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ratpack</groupId>
+            <artifactId>ratpack-test</artifactId>
+            <version>${ratpack.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+            <version>${h2.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>${httpclient.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+            <version>${httpcore.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <argLine>
+                        --add-opens java.base/java.lang=ALL-UNNAMED
+                    </argLine>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <properties>
+        <ratpack.version>1.9.0</ratpack.version>
+        <httpclient.version>4.5.3</httpclient.version>
+        <httpcore.version>4.4.6</httpcore.version>
+        <hystrix.version>1.5.12</hystrix.version>
+        <groovy.sql.version>2.4.15</groovy.sql.version>
+        <ratpack.test.latest.version>1.6.1</ratpack.test.latest.version>
+    </properties>
+
+</project>
@@ -0,0 +1,82 @@
+package com.baeldung;
+
+import com.baeldung.filter.RequestValidatorFilter;
+import com.baeldung.handler.EmployeeHandler;
+import com.baeldung.handler.RedirectHandler;
+import com.baeldung.model.Employee;
+import com.baeldung.repository.EmployeeRepository;
+import com.baeldung.repository.EmployeeRepositoryImpl;
+import com.zaxxer.hikari.HikariConfig;
+import io.netty.buffer.PooledByteBufAllocator;
+import ratpack.exec.internal.DefaultExecController;
+import ratpack.func.Action;
+import ratpack.func.Function;
+import ratpack.guice.BindingsSpec;
+import ratpack.guice.Guice;
+import ratpack.handling.Chain;
+import ratpack.hikari.HikariModule;
+import ratpack.http.client.HttpClient;
+import ratpack.jackson.Jackson;
+import ratpack.registry.Registry;
+import ratpack.server.RatpackServer;
+import ratpack.server.RatpackServerSpec;
+import ratpack.server.ServerConfig;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.List;
+
+public class Application {
+
+    public static void main(String[] args) throws Exception {
+
+        final Action<HikariConfig> hikariConfigAction = hikariConfig -> {
+            hikariConfig.setDataSourceClassName("org.h2.jdbcx.JdbcDataSource");
+            hikariConfig.addDataSourceProperty("URL", "jdbc:h2:mem:baeldung;INIT=RUNSCRIPT FROM 'classpath:/DDL.sql'");
+        };
+
+        final Action<BindingsSpec> bindingsSpecAction = bindings -> bindings.module(HikariModule.class, hikariConfigAction);
+        final HttpClient httpClient = HttpClient.of(httpClientSpec -> {
+            httpClientSpec.poolSize(10)
+                .connectTimeout(Duration.of(60, ChronoUnit.SECONDS))
+                .maxContentLength(ServerConfig.DEFAULT_MAX_CONTENT_LENGTH)
+                .responseMaxChunkSize(16384)
+                .readTimeout(Duration.of(60, ChronoUnit.SECONDS))
+                .byteBufAllocator(PooledByteBufAllocator.DEFAULT).execController(new DefaultExecController(2));
+        });
+        final Function<Registry, Registry> registryFunction = Guice.registry(bindingsSpecAction);
+
+        final Action<Chain> chainAction = chain -> chain.all(new RequestValidatorFilter())
+            .get(ctx -> ctx.render("Welcome to baeldung ratpack!!!"))
+            .get("data/employees", ctx -> ctx.render(Jackson.json(createEmpList())))
+            .get(":name", ctx -> ctx.render("Hello " + ctx.getPathTokens()
+                .get("name") + "!!!"))
+            .post(":amount", ctx -> ctx.render(" Amount $" + ctx.getPathTokens()
+                .get("amount") + " added successfully !!!"));
+
+        final Action<Chain> routerChainAction = routerChain -> {
+            routerChain.path("redirect", new RedirectHandler())
+                .prefix("employee", empChain -> {
+                    empChain.get(":id", new EmployeeHandler());
+                });
+        };
+        final Action<RatpackServerSpec> ratpackServerSpecAction = serverSpec -> serverSpec.registry(registryFunction)
+            .registryOf(registrySpec -> {
+                registrySpec.add(EmployeeRepository.class, new EmployeeRepositoryImpl());
+                registrySpec.add(HttpClient.class, httpClient);
+            })
+            .handlers(chain -> chain.insert(routerChainAction)
+                .insert(chainAction));
+
+        RatpackServer.start(ratpackServerSpecAction);
+    }
+
+    private static List<Employee> createEmpList() {
+        List<Employee> employees = new ArrayList<>();
+        employees.add(new Employee(1L, "Mr", "John Doe"));
+        employees.add(new Employee(2L, "Mr", "White Snow"));
+        return employees;
+    }
+
+}
@@ -0,0 +1,107 @@
+package com.baeldung.model;
+
+import java.time.Instant;
+
+public class Quote {
+    
+    private Instant ts;
+    private String symbol;
+    private double value;
+    
+    public Quote() {}
+    
+    
+    public Quote(Instant ts, String symbol, double value) {
+        this.ts = ts;
+        this.symbol = symbol;
+        this.value = value;
+    }
+
+
+    /**
+     * @return the ts
+     */
+    public Instant getTs() {
+        return ts;
+    }
+    
+    /**
+     * @param ts the ts to set
+     */
+    public void setTs(Instant ts) {
+        this.ts = ts;
+    }
+    
+    /**
+     * @return the symbol
+     */
+    public String getSymbol() {
+        return symbol;
+    }
+    
+    /**
+     * @param symbol the symbol to set
+     */
+    public void setSymbol(String symbol) {
+        this.symbol = symbol;
+    }
+    
+    /**
+     * @return the value
+     */
+    public double getValue() {
+        return value;
+    }
+    
+    /**
+     * @param value the value to set
+     */
+    public void setValue(double value) {
+        this.value = value;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((symbol == null) ? 0 : symbol.hashCode());
+        result = prime * result + ((ts == null) ? 0 : ts.hashCode());
+        long temp;
+        temp = Double.doubleToLongBits(value);
+        result = prime * result + (int) (temp ^ (temp >>> 32));
+        return result;
+    }
+
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        Quote other = (Quote) obj;
+        if (symbol == null) {
+            if (other.symbol != null)
+                return false;
+        } else if (!symbol.equals(other.symbol))
+            return false;
+        if (ts == null) {
+            if (other.ts != null)
+                return false;
+        } else if (!ts.equals(other.ts))
+            return false;
+        if (Double.doubleToLongBits(value) != Double.doubleToLongBits(other.value))
+            return false;
+        return true;
+    }
+
+
+    @Override
+    public String toString() {
+        return "Quote [ts=" + ts + ", symbol=" + symbol + ", value=" + value + "]";
+    }
+    
+    
+}
@@ -0,0 +1,10 @@
+package com.baeldung.repository;
+
+import com.baeldung.model.Employee;
+import ratpack.exec.Promise;
+
+public interface EmployeeRepository {
+
+    Promise<Employee> findEmployeeById(Long id) throws Exception;
+
+}
@@ -0,0 +1,44 @@
+package com.baeldung.rxjava.service;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Random;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.reactivestreams.Publisher;
+
+import com.baeldung.model.Quote;
+
+import ratpack.stream.Streams;
+
+public class QuotesService {
+
+    private final ScheduledExecutorService executorService;
+    private static Random rnd = new Random(); 
+    private static String[] symbols = new String[] {
+      "MSFT",
+      "ORCL",
+      "GOOG",
+      "AAPL",
+      "CSCO"
+    };
+    
+    public QuotesService(ScheduledExecutorService executorService) {
+        this.executorService = executorService;
+    }
+
+    public Publisher<Quote> newTicker() {
+        return Streams.periodically(executorService, Duration.ofSeconds(2), (t) -> {
+           
+            return randomQuote();
+        });
+    }
+    
+    private static Quote randomQuote() {
+        return new Quote (
+            Instant.now(),
+            symbols[rnd.nextInt(symbols.length)],
+            Math.round(rnd.nextDouble()*100)
+        );
+    }
+}
@@ -0,0 +1,206 @@
+package com.baeldung.spring;
+
+import java.util.Random;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+import com.baeldung.model.Quote;
+import com.baeldung.rxjava.service.QuotesService;
+
+import groovy.util.logging.Slf4j;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import ratpack.func.Action;
+import ratpack.handling.Chain;
+import ratpack.http.ResponseChunks;
+import ratpack.http.Status;
+import ratpack.server.ServerConfig;
+import ratpack.spring.config.EnableRatpack;
+import ratpack.sse.ServerSentEvents;
+import ratpack.stream.Streams;
+import ratpack.stream.TransformablePublisher;
+import ratpack.websocket.WebSockets;
+import rx.subscriptions.Subscriptions;
+
+/**
+ * @author psevestre
+ */
+@SpringBootApplication
+@EnableRatpack
+public class EmbedRatpackStreamsApp {
+    
+    private static final Logger log = LoggerFactory.getLogger(EmbedRatpackStreamsApp.class);
+
+    @Autowired 
+    private QuotesService quotesService;
+    
+    private AtomicLong idSeq =  new AtomicLong(0);
+    
+    
+    @Bean
+    public ScheduledExecutorService executorService() {
+        return Executors.newScheduledThreadPool(1);
+    }
+    
+    @Bean
+    public QuotesService quotesService(ScheduledExecutorService executor) {
+        return new QuotesService(executor);
+    }
+
+    @Bean
+    public Action<Chain> quotes() {
+        ServerSentEvents sse = ServerSentEvents.serverSentEvents(quotesService.newTicker(), (evt) -> {
+            evt
+              .id(Long.toString(idSeq.incrementAndGet()))
+              .event("quote")
+              .data( q -> q.toString());
+        });
+        
+        return chain -> chain.get("quotes", ctx -> ctx.render(sse));
+    }
+
+    @Bean
+    public Action<Chain> quotesWS() {
+        Publisher<String> pub = Streams.transformable(quotesService.newTicker())
+          .map(Quote::toString);
+        return chain -> chain.get("quotes-ws", ctx -> WebSockets.websocketBroadcast(ctx, pub));
+    }
+    
+    @Bean
+    public Action<Chain> uploadFile() {
+        
+        return chain -> chain.post("upload", ctx -> {
+            TransformablePublisher<? extends ByteBuf> pub = ctx.getRequest().getBodyStream();
+            pub.subscribe(new Subscriber<ByteBuf>() {
+                private Subscription sub;
+                @Override
+                public void onSubscribe(Subscription sub) {
+                    this.sub = sub;
+                    sub.request(1);
+                }
+    
+                @Override
+                public void onNext(ByteBuf t) {
+                    try {
+                        int len = t.readableBytes();
+                        log.info("Got {} bytes", len);
+    
+                        // Do something useful with data
+                        
+                        // Request next chunk
+                        sub.request(1);
+                    }
+                    finally {
+                        // DO NOT FORGET to RELEASE !
+                        t.release();
+                    }
+                }
+    
+                @Override
+                public void onError(Throwable t) {
+                    ctx.getResponse().status(500);
+                }
+    
+                @Override
+                public void onComplete() {
+                    ctx.getResponse().status(202);
+                }
+            }); 
+        });
+    }
+    
+    @Bean
+    public Action<Chain> download() {
+        return chain -> chain.get("download", ctx -> {
+            ctx.getResponse().sendStream(new RandomBytesPublisher(1024,512));
+        });
+    }
+
+    @Bean
+    public Action<Chain> downloadChunks() {
+        return chain -> chain.get("downloadChunks", ctx -> {
+            ctx.render(ResponseChunks.bufferChunks("application/octetstream",
+              new RandomBytesPublisher(1024,512)));
+        });
+    }
+    
+    @Bean
+    public ServerConfig ratpackServerConfig() {
+        return ServerConfig
+          .builder()
+          .findBaseDir("public")
+          .build();
+    }
+
+    public static void main(String[] args) {
+        SpringApplication.run(EmbedRatpackStreamsApp.class, args);
+    }
+    
+    
+    public static class RandomBytesPublisher implements Publisher<ByteBuf> {
+        
+        private int bufCount;
+        private int bufSize;
+        private Random rnd = new Random();
+        
+
+        RandomBytesPublisher(int bufCount, int bufSize) {
+            this.bufCount = bufCount;
+            this.bufSize = bufSize;
+        }
+
+        @Override
+        public void subscribe(Subscriber<? super ByteBuf> s) {
+            s.onSubscribe(new Subscription() {
+                
+                private boolean cancelled = false;
+                private boolean recurse;
+                private long requested = 0;
+                
+                @Override
+                public void request(long n) {
+                    if ( bufCount == 0 ) {
+                        s.onComplete();
+                        return;
+                    }
+                    
+                    requested += n;
+                    if ( recurse ) {
+                       return; 
+                    }
+                    
+                    recurse = true;
+                    try {
+                        while ( requested-- > 0 && !cancelled && bufCount-- > 0 ) {
+                            byte[] data = new byte[bufSize];
+                            rnd.nextBytes(data);
+                            ByteBuf buf = Unpooled.wrappedBuffer(data);
+                            s.onNext(buf);
+                        }
+                    }
+                    finally {
+                        recurse = false;
+                    }
+                }
+                
+                @Override
+                public void cancel() {
+                    cancelled = true;
+                }
+            });
+            
+        }
+    }
+
+}
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
+            </pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>
@@ -0,0 +1,63 @@
+package com.baeldung.ratpack;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Non-thread safe !!!
+class CompliantPublisher implements Publisher<Integer> {
+    String name;
+
+    private static final Logger log = LoggerFactory.getLogger(CompliantPublisher.class); 
+    private long available;
+
+
+    public CompliantPublisher(long available) {
+        this.available = available;
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super Integer> subscriber) {
+        log.info("subscribe");
+        subscriber.onSubscribe(new CompliantSubscription(subscriber));
+        
+    }
+    
+    
+    private class CompliantSubscription implements Subscription {
+
+        private Subscriber<? super Integer> subscriber;
+        private int recurseLevel;
+        private long requested;
+        private boolean cancelled;
+
+        public CompliantSubscription(Subscriber<? super Integer> subscriber) {
+            this.subscriber = subscriber;
+        }
+
+        @Override
+        public void request(long n) {
+            log.info("request: requested={}, available={}", n, available);
+            requested += n;
+            if ( recurseLevel > 0 ) {
+               return;
+            }
+            
+            recurseLevel++;
+            for (int i = 0 ; i < (requested) && !cancelled && available > 0 ; i++, available-- ) {
+                subscriber.onNext(i);
+            }
+            subscriber.onComplete();
+        }
+
+        @Override
+        public void cancel() {
+            cancelled  = true;
+        }
+        
+    }
+    
+}
@@ -0,0 +1,67 @@
+package com.baeldung.ratpack;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class LoggingSubscriber<T> implements Subscriber<T> {
+    private static final Logger log = LoggerFactory.getLogger(LoggingSubscriber.class);
+
+    private Subscription subscription;
+    private long requested;
+    private long received;
+    private CountDownLatch finished = new CountDownLatch(1);
+
+    @Override
+    public void onComplete() {
+        log.info("onComplete: sub={}", subscription.hashCode());
+        finished.countDown();
+    }
+
+    @Override
+    public void onError(Throwable t) {
+        log.error("Error: sub={}, message={}", subscription.hashCode(), t.getMessage(),t);
+        finished.countDown();
+    }
+
+    @Override
+    public void onNext(T value) {
+        log.info("onNext: sub={}, value={}", subscription.hashCode(), value);
+        this.received++;
+        this.requested++;
+        subscription.request(1);
+    }
+
+    @Override
+    public void onSubscribe(Subscription sub) {
+        log.info("onSubscribe: sub={}", sub.hashCode());
+        this.subscription = sub;
+        this.received = 0;
+        this.requested = 1;
+        sub.request(1);
+    }
+    
+    
+    public long getRequested() {
+        return requested;
+    }
+    
+    public long getReceived() {
+        return received;
+    }
+
+    public void block() {
+        try {
+            finished.await(10, TimeUnit.SECONDS);
+        }
+        catch(InterruptedException iex) {
+            throw new RuntimeException(iex);
+        }
+    }
+
+}
@@ -0,0 +1,46 @@
+package com.baeldung.ratpack;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class NonCompliantPublisher implements Publisher<Integer> {
+    
+    private static final Logger log = LoggerFactory.getLogger(NonCompliantPublisher.class);
+
+    @Override
+    public void subscribe(Subscriber<? super Integer> subscriber) {
+        log.info("subscribe");
+        subscriber.onSubscribe(new NonCompliantSubscription(subscriber));
+    }
+    
+    private class NonCompliantSubscription implements Subscription {
+        private Subscriber<? super Integer> subscriber;
+        private int recurseLevel = 0;
+
+        public NonCompliantSubscription(Subscriber<? super Integer> subscriber) {
+            this.subscriber = subscriber;
+        }
+
+        @Override
+        public void request(long n) {
+            log.info("request: n={}", n);
+            if ( recurseLevel > 0 ) {
+               return;
+            }
+            
+            recurseLevel++;
+            for (int i = 0 ; i < (n + 5) ; i ++ ) {
+                subscriber.onNext(i);
+            }
+            subscriber.onComplete();
+        }
+
+        @Override
+        public void cancel() {
+        }
+    }
+}
@@ -0,0 +1,140 @@
+package com.baeldung.ratpack;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import ratpack.exec.ExecResult;
+import ratpack.func.Action;
+import ratpack.stream.StreamEvent;
+import ratpack.stream.Streams;
+import ratpack.stream.TransformablePublisher;
+import ratpack.test.exec.ExecHarness;
+
+public class RatpackStreamsUnitTest {
+    
+    private static Logger log = LoggerFactory.getLogger(RatpackStreamsUnitTest.class);
+
+    @Test
+    public void whenPublish_thenSuccess() {
+        
+        Publisher<String> pub = Streams.publish(Arrays.asList("hello", "hello again"));
+        LoggingSubscriber<String> sub = new LoggingSubscriber<String>();
+        pub.subscribe(sub);
+        sub.block();
+    }
+    
+    
+    @Test
+    public void whenYield_thenSuccess() {
+        
+        Publisher<String> pub = Streams.yield((t) -> {
+            return t.getRequestNum() < 5 ? "hello" : null;
+        });
+        
+        LoggingSubscriber<String> sub = new LoggingSubscriber<String>();
+        pub.subscribe(sub);
+        sub.block();
+        assertEquals(5, sub.getReceived());
+    }
+    
+    @Test
+    public void whenPeriodic_thenSuccess() {
+        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+        Publisher<String> pub = Streams.periodically(executor, Duration.ofSeconds(1), (t) -> {
+            return t < 5 ? String.format("hello %d",t): null; 
+        });
+
+        LoggingSubscriber<String> sub = new LoggingSubscriber<String>();
+        pub.subscribe(sub);
+        sub.block();
+        assertEquals(5, sub.getReceived());
+    }
+    
+    @Test
+    public void whenMap_thenSuccess() throws Exception {
+        
+        TransformablePublisher<String> pub = Streams.yield( t -> {
+            return t.getRequestNum() < 5 ? t.getRequestNum() : null;
+          })
+          .map(v -> String.format("item %d", v));
+        
+        ExecResult<List<String>> result = ExecHarness.yieldSingle((c) -> pub.toList() );
+        assertTrue("should succeed", result.isSuccess());
+        assertEquals("should have 5 items",5,result.getValue().size());
+    }
+    
+    @Test
+    public void whenNonCompliantPublisherWithBuffer_thenSuccess() throws Exception {
+        
+        TransformablePublisher<Integer> pub = Streams.transformable(new NonCompliantPublisher())
+          .wiretap(new LoggingAction("before buffer"))
+          .buffer()
+          .wiretap(new LoggingAction("after buffer"))
+          .take(1);
+          
+        LoggingSubscriber<Integer> sub = new LoggingSubscriber<>();
+        pub.subscribe(sub);
+        sub.block();
+    }
+    
+    @Test
+    public void whenNonCompliantPublisherWithoutBuffer_thenSuccess() throws Exception {
+        TransformablePublisher<Integer> pub = Streams.transformable(new NonCompliantPublisher())
+          .wiretap(new LoggingAction(""))
+          .take(1);
+          
+        LoggingSubscriber<Integer> sub = new LoggingSubscriber<>();
+        pub.subscribe(sub);
+        sub.block();
+    }
+
+@Test
+public void whenCompliantPublisherWithoutBatch_thenSuccess() throws Exception {
+    
+    TransformablePublisher<Integer> pub = Streams.transformable(new CompliantPublisher(10))
+      .wiretap(new LoggingAction(""));
+      
+    LoggingSubscriber<Integer> sub = new LoggingSubscriber<>();
+    pub.subscribe(sub);
+    sub.block();
+}
+
+@Test
+public void whenCompliantPublisherWithBatch_thenSuccess() throws Exception {
+    
+    TransformablePublisher<Integer> pub = Streams.transformable(new CompliantPublisher(10))
+      .wiretap(new LoggingAction("before batch"))
+      .batch(5, Action.noop())
+      .wiretap(new LoggingAction("after batch"));
+      
+    LoggingSubscriber<Integer> sub = new LoggingSubscriber<>();
+    pub.subscribe(sub);
+    sub.block();
+}
+    
+    private static class LoggingAction implements Action<StreamEvent<Integer>>{
+        private final String label;
+
+        public LoggingAction(String label) {
+            this.label = label;
+        }
+
+        @Override
+        public void execute(StreamEvent<Integer> e) throws Exception {
+            log.info("{}: event={}", label,e);
+        }
+        
+    }
+    
+}