-
Notifications
You must be signed in to change notification settings - Fork 596
/
package.scala
179 lines (170 loc) · 7.11 KB
/
package.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
/*
* Copyright (c) 2013 Functional Streams for Scala
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package fs2
package interop
import cats.effect.kernel.{Async, Resource}
import java.util.concurrent.Flow.{Publisher, Subscriber, defaultBufferSize}
/** Implementation of the reactive-streams protocol for fs2; based on Java Flow.
*
* @example {{{
* scala> import cats.effect.{IO, Resource}
* scala> import fs2.Stream
* scala> import fs2.interop.flow.syntax._
* scala> import java.util.concurrent.Flow.Publisher
* scala>
* scala> val upstream: Stream[IO, Int] = Stream(1, 2, 3).covary[IO]
* scala> val publisher: Resource[IO, Publisher[Int]] = upstream.toPublisher
* scala> val downstream: Stream[IO, Int] = Stream.resource(publisher).flatMap { publisher =>
* | publisher.toStream[IO](chunkSize = 16)
* | }
* scala>
* scala> import cats.effect.unsafe.implicits.global
* scala> downstream.compile.toVector.unsafeRunSync()
* res0: Vector[Int] = Vector(1, 2, 3)
* }}}
*
* @see [[java.util.concurrent.Flow]]
*/
package object flow {
/** Creates a [[Stream]] from a `subscribe` function;
* analogous to a `Publisher`, but effectual.
*
* This function is useful when you actually need to provide a subscriber to a third-party.
*
* @example {{{
* scala> import cats.effect.IO
* scala> import fs2.Stream
* scala> import java.util.concurrent.Flow.{Publisher, Subscriber}
* scala>
* scala> def thirdPartyLibrary(subscriber: Subscriber[Int]): Unit = {
* | def somePublisher: Publisher[Int] = ???
* | somePublisher.subscribe(subscriber)
* | }
* scala>
* scala> // Interop with the third party library.
* scala> fs2.interop.flow.fromPublisher[IO, Int](chunkSize = 16) { subscriber =>
* | IO.println("Subscribing!") >>
* | IO.delay(thirdPartyLibrary(subscriber)) >>
* | IO.println("Subscribed!")
* | }
* res0: Stream[IO, Int] = Stream(..)
* }}}
*
* @note The subscribe function will not be executed until the stream is run.
*
* @see the overload that only requires a [[Publisher]].
*
* @param chunkSize setup the number of elements asked each time from the [[Publisher]].
* A high number may be useful if the publisher is triggering from IO,
* like requesting elements from a database.
* A high number will also lead to more elements in memory.
* The stream will not emit new element until,
* either the `Chunk` is filled or the publisher finishes.
* @param subscribe The effectual function that will be used to initiate the consumption process,
* it receives a [[Subscriber]] that should be used to subscribe to a [[Publisher]].
* The `subscribe` operation must be called exactly once.
*/
def fromPublisher[F[_], A](
chunkSize: Int
)(
subscribe: Subscriber[A] => F[Unit]
)(implicit
F: Async[F]
): Stream[F, A] =
Stream
.eval(StreamSubscriber[F, A](chunkSize))
.flatMap { subscriber =>
subscriber.stream(subscribe(subscriber))
}
/** Creates a [[Stream]] from an [[Publisher]].
*
* @example {{{
* scala> import cats.effect.IO
* scala> import fs2.Stream
* scala> import java.util.concurrent.Flow.Publisher
* scala>
* scala> def getThirdPartyPublisher(): Publisher[Int] = ???
* scala>
* scala> // Interop with the third party library.
* scala> Stream.eval(IO.delay(getThirdPartyPublisher())).flatMap { publisher =>
* | fs2.interop.flow.fromPublisher[IO](publisher, chunkSize = 16)
* | }
* res0: Stream[IO, Int] = Stream(..)
* }}}
*
* @note The publisher will not receive a subscriber until the stream is run.
*
* @see the `toStream` extension method added to `Publisher`
*
* @param publisher The [[Publisher]] to consume.
* @param chunkSize setup the number of elements asked each time from the [[Publisher]].
* A high number may be useful if the publisher is triggering from IO,
* like requesting elements from a database.
* A high number will also lead to more elements in memory.
* The stream will not emit new element until,
* either the `Chunk` is filled or the publisher finishes.
*/
def fromPublisher[F[_]]: syntax.FromPublisherPartiallyApplied[F] =
new syntax.FromPublisherPartiallyApplied(dummy = true)
/** Creates a [[Publisher]] from a [[Stream]].
*
* The stream is only ran when elements are requested.
* Closing the [[Resource]] means gracefully shutting down all active subscriptions.
* Thus, no more elements will be published.
*
* @note This Publisher can be reused for multiple Subscribers,
* each subscription will re-run the [[Stream]] from the beginning.
*
* @see [[subscribeStream]] for a simpler version that only requires a [[Subscriber]].
*
* @param stream The [[Stream]] to transform.
*/
def toPublisher[F[_], A](
stream: Stream[F, A]
)(implicit
F: Async[F]
): Resource[F, Publisher[A]] =
StreamPublisher(stream)
/** Allows subscribing a [[Subscriber]] to a [[Stream]].
*
* The returned program will run until
* all the stream elements were consumed.
* Cancelling this program will gracefully shutdown the subscription.
*
* @param stream the [[Stream]] that will be consumed by the subscriber.
* @param subscriber the [[Subscriber]] that will receive the elements of the stream.
*/
def subscribeStream[F[_], A](
stream: Stream[F, A],
subscriber: Subscriber[A]
)(implicit
F: Async[F]
): F[Unit] =
StreamSubscription.subscribe(stream, subscriber)
/** A default value for the `chunkSize` argument,
* that may be used in the absence of other constraints;
* we encourage choosing an appropriate value consciously.
* Alias for [[defaultBufferSize]].
*
* @note the current value is `256`.
*/
val defaultChunkSize: Int = defaultBufferSize
}