Skip to content

Commit 289ab12

Browse files
akarnokdakarnokd
committed
Improve Javadoc, fix headers, add proper checkstyle
1 parent b22ec28 commit 289ab12

23 files changed

+213
-29
lines changed

‎HEADER

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
Copyright 2019 David Karnok
1+
Copyright 2019-Present David Karnok
22

33
Licensed under the Apache License, Version 2.0 (the "License");
44
you may not use this file except in compliance with the License.

‎config/checkstyle/checkstyle.xml

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
<?xml version="1.0"?>
2+
<!DOCTYPE module PUBLIC
3+
"-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
4+
"https://checkstyle.org/dtds/configuration_1_3.dtd">
5+
6+
<module name="Checker">
7+
<module name="SuppressionFilter">
8+
<property name="file" value="${checkstyle.suppressions.file}"/>
9+
</module>
10+
11+
<!-- Headers -->
12+
<module name="Header">
13+
<property name="headerFile" value="${checkstyle.header.file}"/>
14+
<property name="fileExtensions" value="java"/>
15+
</module>
16+
17+
<module name="TreeWalker">
18+
<module name="JavadocMethod"/>
19+
<module name="MissingJavadocMethod"/>
20+
21+
<module name="RegexpSinglelineJava">
22+
<property name="severity" value="warning"/>
23+
<property name="format" value="^(?!\s+\* $).*?\s+$"/>
24+
<property name="message" value="Line has trailing spaces."/>
25+
</module>
26+
</module>
27+
28+
</module>

‎config/checkstyle/suppressions.xml

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
<?xml version="1.0"?>
2+
3+
<!DOCTYPE suppressions PUBLIC
4+
"-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN"
5+
"https://checkstyle.org/dtds/suppressions_1_2.dtd">
6+
7+
<suppressions>
8+
9+
<suppress checks="MissingJavadocMethod" files="[\\/]main[\\/]hu[\\/]akarnokd[\\/]rxjava3[\\/]fibers[\\/]internal[\\/]"/>
10+
<suppress checks="MissingJavadocMethod" files="[\\/]jmh[\\/]"/>
11+
<suppress checks="MissingJavadocMethod" files="[\\/]test[\\/]"/>
12+
13+
</suppressions>

‎config/license/HEADER

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
Copyright 2019-Present David Karnok
2+
3+
Licensed under the Apache License, Version 2.0 (the "License");
4+
you may not use this file except in compliance with the License.
5+
You may obtain a copy of the License at
6+
7+
http://www.apache.org/licenses/LICENSE-2.0
8+
9+
Unless required by applicable law or agreed to in writing, software
10+
distributed under the License is distributed on an "AS IS" BASIS,
11+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
See the License for the specific language governing permissions and
13+
limitations under the License.

‎config/license/HEADER_JAVA

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
/*
2+
* Copyright 2019-Present David Karnok
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/

‎src/main/java/hu/akarnokd/rxjava3/fibers/FiberEmitter.java

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,8 +16,17 @@
1616

1717
package hu.akarnokd.rxjava3.fibers;
1818

19+
/**
20+
* Interface handed to user code in {@link FiberInterop#create(FiberGenerator, java.util.concurrent.ExecutorService)} callback.
21+
* @param <T> the element type to emit
22+
*/
1923
@FunctionalInterface
2024
public interface FiberEmitter<T> {
2125

26+
/**
27+
* Signal the next item
28+
* @param item the item to signal
29+
* @throws Throwable an arbitrary exception if the downstream cancelled
30+
*/
2231
void emit(T item) throws Throwable;
2332
}

‎src/main/java/hu/akarnokd/rxjava3/fibers/FiberGenerator.java

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,8 +16,21 @@
1616

1717
package hu.akarnokd.rxjava3.fibers;
1818

19+
/**
20+
* Interface to implement to produce elements when asked by
21+
* {@link FiberInterop#create(FiberGenerator, java.util.concurrent.ExecutorService)}.
22+
* <p>
23+
* To signal {@code onComplete}, return normally from {@link #generate(FiberEmitter)}.
24+
* To signal {@code onError}, throw any exception from {@link #generate(FiberEmitter)}.
25+
* @param <T> the element type generated
26+
*/
1927
@FunctionalInterface
2028
public interface FiberGenerator<T> {
2129

30+
/**
31+
* The method to implement and start emitting items.
32+
* @param emitter use {@link FiberEmitter#emit(Object)} to generate values
33+
* @throws Throwable if the generator wishes to signal {@code onError}.
34+
*/
2235
void generate(FiberEmitter<T> emitter) throws Throwable;
2336
}

‎src/main/java/hu/akarnokd/rxjava3/fibers/FiberInterop.java

Lines changed: 79 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -19,13 +19,34 @@
1919
import java.util.Objects;
2020
import java.util.concurrent.*;
2121

22+
import io.reactivex.rxjava3.annotations.BackpressureKind;
23+
import io.reactivex.rxjava3.annotations.BackpressureSupport;
24+
import io.reactivex.rxjava3.annotations.SchedulerSupport;
2225
import io.reactivex.rxjava3.core.*;
2326
import io.reactivex.rxjava3.internal.functions.ObjectHelper;
2427
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
2528

2629
/**
27-
* Sources, transformers and consumers working with fiber-based suspendable methods.
28-
*
30+
* Sources and transformers working with
31+
* virtual thread-based executors that allow
32+
* efficient blocking via suspensions and resumptions.
33+
* <p>
34+
* Examples:
35+
* <pre><code>
36+
* try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
37+
* FiberInterop.<Integer>create(emitter -> {
38+
* for (int i = 0; i &lt; 10; i++) {
39+
* Thread.sleep(1000);
40+
* emitter.emit(i);
41+
* }
42+
* }, executor)
43+
* .subscribe(
44+
* System.out::println,
45+
* Throwable::printStackTrace,
46+
* () -> System.out.println("Done")
47+
* );
48+
* }
49+
* </code></pre>
2950
* @since 0.0.1
3051
*/
3152
public final class FiberInterop {
@@ -35,20 +56,70 @@ private FiberInterop() {
3556
throw new IllegalStateException("No instances!");
3657
}
3758

59+
/**
60+
* Construct a {@link Flowable} and use the given {@code generator}
61+
* to generate items on demand while running on the given {@link ExecutorService}.
62+
* <p>
63+
* Note that backpressure is handled via blocking so it is recommended the provided
64+
* {@code ExecutorService} uses virtual threads, such as the one returned by
65+
* {@link Executors#newVirtualThreadPerTaskExecutor()}.
66+
* @param <T> the element type to emit
67+
* @param generator the callback used to generate items on demand by the downstream
68+
* @param executor the target {@code ExecutorService} to use for running the callback
69+
* @return the new {@code Flowable} instance
70+
*/
71+
@BackpressureSupport(BackpressureKind.FULL)
72+
@SchedulerSupport(SchedulerSupport.CUSTOM)
73+
@SuppressWarnings("preview")
3874
public static <T> Flowable<T> create(FiberGenerator<T> generator, ExecutorService executor) {
3975
Objects.requireNonNull(generator, "generator is null");
4076
Objects.requireNonNull(executor, "executor is null");
4177
return RxJavaPlugins.onAssembly(new FlowableCreateFiberExecutor<>(generator, executor));
4278
}
4379

44-
public static <T, R> FlowableTransformer<T, R> transform(FiberTransformer<T, R> transformer, ExecutorService scheduler) {
45-
return transform(transformer, scheduler, Flowable.bufferSize());
80+
/**
81+
* Construct a transformer to be used via {@link Flowable#compose(FlowableTransformer)}
82+
* which can turn an upstream item into zero or more downstream values by running
83+
* on the given {@link ExecutorService}.
84+
* <p>
85+
* Note that backpressure is handled via blocking so it is recommended the provided
86+
* {@code ExecutorService} uses virtual threads, such as the one returned by
87+
* {@link Executors#newVirtualThreadPerTaskExecutor()}.
88+
* @param <T> the upstream element type
89+
* @param <R> the downstream element type
90+
* @param transformer the callback whose {@link FiberTransformer#transform(Object, FiberEmitter)} is invoked for each upstream item
91+
* @param executor the target {@code ExecutorService} to use for running the callback
92+
* @return the new {@code FlowableTransformer} instance
93+
*/
94+
@BackpressureSupport(BackpressureKind.FULL)
95+
@SchedulerSupport(SchedulerSupport.CUSTOM)
96+
@SuppressWarnings("preview")
97+
public static <T, R> FlowableTransformer<T, R> transform(FiberTransformer<T, R> transformer, ExecutorService executor) {
98+
return transform(transformer, executor, Flowable.bufferSize());
4699
}
47100

48-
public static <T, R> FlowableTransformer<T, R> transform(FiberTransformer<T, R> transformer, ExecutorService scheduler, int prefetch) {
101+
/**
102+
* Construct a transformer to be used via {@link Flowable#compose(FlowableTransformer)}
103+
* which can turn an upstream item into zero or more downstream values by running
104+
* on the given {@link ExecutorService}.
105+
* <p>
106+
* Note that backpressure is handled via blocking so it is recommended the provided
107+
* {@code ExecutorService} uses virtual threads, such as the one returned by
108+
* {@link Executors#newVirtualThreadPerTaskExecutor()}.
109+
* @param <T> the upstream element type
110+
* @param <R> the downstream element type
111+
* @param transformer the callback whose {@link FiberTransformer#transform(Object, FiberEmitter)} is invoked for each upstream item
112+
* @param executor the target {@code ExecutorService} to use for running the callback
113+
* @param prefetch the number of items to fetch from the upstream.
114+
* @return the new {@code FlowableTransformer} instance
115+
*/
116+
@BackpressureSupport(BackpressureKind.FULL)
117+
@SchedulerSupport(SchedulerSupport.CUSTOM)
118+
@SuppressWarnings("preview")
119+
public static <T, R> FlowableTransformer<T, R> transform(FiberTransformer<T, R> transformer, ExecutorService executor, int prefetch) {
49120
Objects.requireNonNull(transformer, "transformer is null");
50-
Objects.requireNonNull(scheduler, "scheduler is null");
121+
Objects.requireNonNull(executor, "executor is null");
51122
ObjectHelper.verifyPositive(prefetch, "prefetch");
52-
return new FlowableTransformFiberExecutor<>(null, transformer, scheduler, prefetch);
123+
return new FlowableTransformFiberExecutor<>(null, transformer, executor, prefetch);
53124
}
54125
}

‎src/main/java/hu/akarnokd/rxjava3/fibers/FiberTransformer.java

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,8 +16,23 @@
1616

1717
package hu.akarnokd.rxjava3.fibers;
1818

19+
/**
20+
* Interface called by the {@link FiberInterop#transform(FiberTransformer, java.util.concurrent.ExecutorService)}
21+
* operator to generate any number of output values based of the current input of the upstream.
22+
*
23+
* @param <T> the source value type
24+
* @param <R> the result value type
25+
*/
1926
@FunctionalInterface
2027
public interface FiberTransformer<T, R> {
2128

29+
/**
30+
* Implement this method to generate any number of items via
31+
* {@link FiberEmitter#emit(Object)}.
32+
*
33+
* @param value the upstream value
34+
* @param emitter the emitter to use to generate result value(s)
35+
* @throws Throwable signaled as {@code onError} for the downstream.
36+
*/
2237
void transform(T value, FiberEmitter<R> emitter) throws Throwable;
2338
}

‎src/main/java/hu/akarnokd/rxjava3/fibers/FlowableCreateFiberExecutor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/fibers/FlowableTransformFiberExecutor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/fibers/ResumableFiber.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
2121

2222
/**
2323
* Fundamental primitive for suspending and resuming a Thread.
24+
* @since 0.0.1
2425
*/
2526
public class ResumableFiber extends AtomicReference<Object> {
2627

@@ -46,7 +47,7 @@ public final void await() {
4647
}
4748

4849
if (current != null && current != toUnpark) {
49-
throw new IllegalStateException("Only one Thread/Fiber can await this ResumableFiber!");
50+
throw new IllegalStateException("Only one (Virtual)Thread can await this ResumableFiber!");
5051
}
5152

5253
if (compareAndSet(null, toUnpark)) {

‎src/test/java/hu/akarnokd/rxjava3/fibers/FiberInteropTest.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@ public void checkClass() {
3434
}
3535

3636
@Test
37+
@SuppressWarnings("preview")
3738
public void checkIsInsideFiber() {
3839
try (var scope = Executors.newVirtualThreadPerTaskExecutor()) {
3940
FiberInterop.create(emitter -> {
@@ -46,6 +47,7 @@ public void checkIsInsideFiber() {
4647
}
4748

4849
@Test
50+
@SuppressWarnings("preview")
4951
public void checkIsInsideFiberExec() throws Throwable {
5052
try (var exec = Executors.newSingleThreadExecutor()) {
5153
FiberInterop.create(emitter -> {
@@ -60,6 +62,7 @@ public void checkIsInsideFiberExec() throws Throwable {
6062
}
6163

6264
@Test
65+
@SuppressWarnings("preview")
6366
public void plainVirtual() {
6467
var result = new AtomicReference<Boolean>();
6568
try (var scope = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory())) {
@@ -69,6 +72,7 @@ public void plainVirtual() {
6972
assertTrue(result.get());
7073
}
7174

75+
@SuppressWarnings("preview")
7276
static void withVirtual(Consumer<ExecutorService> call) throws Throwable {
7377
try (var exec = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory())) {
7478
call.accept(exec);

‎src/test/java/hu/akarnokd/rxjava3/fibers/ResumableFiberTest.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -66,6 +66,7 @@ public void concurrentAwait() {
6666
}
6767

6868
@Test(timeOut = 30000)
69+
@SuppressWarnings("preview")
6970
public void pingPong() throws Exception {
7071
try (var scope = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory())) {
7172

‎src/test/java/hu/akarnokd/rxjava3/fibers/SetupLatestLoom.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -47,7 +47,7 @@ public static void main(String[] args) throws Exception {
4747
int k = s.indexOf("\"", i + 10);
4848
url = s.substring(i, k);
4949

50-
if (url.endsWith(".tar.gz") && url.contains("linux")) {
50+
if (url.endsWith(".tar.gz") && url.contains("linux-x64")) {
5151
break;
5252
}
5353
url = "";

‎src/test/java/hu/akarnokd/rxjava3/fibers/TestHelper.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019 David Karnok
2+
* Copyright 2019-Present David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

0 commit comments

Comments
 (0)