Skip to content

Commit e2d3fe9

Browse files
teivahteivah
committed
Defer operator
1 parent 2640fee commit e2d3fe9

File tree

4 files changed

+67
-67
lines changed

4 files changed

+67
-67
lines changed

‎README.md

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -185,10 +185,10 @@ The result of this execution is:
185185

186186
It means, the first Observer consumed already all the items.
187187

188-
On the other hand, let's create a hot Observable using `FromFuncs` operator:
188+
On the other hand, let's create a hot Observable using `Defer` operator:
189189

190190
```go
191-
observable := rxgo.FromFuncs(func(_ context.Context, ch chan<- rxgo.Item, done func()) {
191+
observable := rxgo.Defer(func(_ context.Context, ch chan<- rxgo.Item, done func()) {
192192
for i := 0; i < 3; i++ {
193193
ch <- rxgo.Of(i)
194194
}
@@ -217,7 +217,7 @@ Now, the result is:
217217
2
218218
```
219219

220-
In the case of a hot observable created with `FromFuncs`, the stream is reproducible. Depending on our use case, we may favour one or the other approach.
220+
In the case of a hot observable created with `Defer`, the stream is reproducible. Depending on our use case, we may favour one or the other approach.
221221

222222
### Backpressure
223223

@@ -267,7 +267,7 @@ In this example, we create a pool of 32 goroutines that consume items concurrent
267267
* [Empty/Never](http://reactivex.io/documentation/operators/empty-never-throw.html) — create Observables that have very precise and limited behaviour
268268
* FromChannel — create an Observable based on a lazy channel
269269
* FromEventSource — create an Observable based on an eager channel
270-
* FromFuncs - combine scatter functions emitting items into one Observable
270+
* Defer - combine scatter functions emitting items into one Observable
271271
* FromSlice — create an Observable from a slice
272272
* [Interval](http://reactivex.io/documentation/operators/interval.html) — create an Observable that emits a sequence of integers spaced by a particular time interval
273273
* [Just](http://reactivex.io/documentation/operators/just.html) — convert a set of objects into an Observable that emits that or those objects

‎factory.go

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,13 @@ func Concat(observables []Observable, opts ...Option) Observable {
162162
}
163163
}
164164

165+
// Defer creates an observable from multiple functions.
166+
func Defer(f ...ProducerFunc) Observable {
167+
return &observable{
168+
iterable: newFuncsIterable(f...),
169+
}
170+
}
171+
165172
// Empty creates an Observable with no item and terminate immediately.
166173
func Empty() Observable {
167174
next := make(chan Item)
@@ -300,13 +307,6 @@ func Range(start, count int) Observable {
300307
}
301308
}
302309

303-
// FromFuncs creates an observable from multiple functions.
304-
func FromFuncs(f ...ProducerFunc) Observable {
305-
return &observable{
306-
iterable: newFuncsIterable(f...),
307-
}
308-
}
309-
310310
// Start creates an Observable from one or more directive-like Supplier
311311
// and emits the result of each operation asynchronously on a new Observable.
312312
func Start(fs []Supplier, opts ...Option) Observable {

‎factory_test.go

Lines changed: 54 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -80,43 +80,8 @@ func Test_Concat_OneEmptyObservable(t *testing.T) {
8080
Assert(context.Background(), t, obs, HasItems(1, 2, 3))
8181
}
8282

83-
func Test_Empty(t *testing.T) {
84-
obs := Empty()
85-
Assert(context.Background(), t, obs, HasNoItems())
86-
}
87-
88-
func Test_FromChannel(t *testing.T) {
89-
ch := make(chan Item)
90-
go func() {
91-
ch <- Of(1)
92-
ch <- Of(2)
93-
ch <- Of(3)
94-
close(ch)
95-
}()
96-
obs := FromChannel(ch)
97-
Assert(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
98-
}
99-
100-
func Test_FromChannel_SimpleCapacity(t *testing.T) {
101-
ch := FromChannel(make(chan Item, 10)).Observe(WithBufferedChannel(11))
102-
assert.Equal(t, 10, cap(ch))
103-
}
104-
105-
func Test_FromChannel_ComposedCapacity(t *testing.T) {
106-
obs1 := FromChannel(make(chan Item, 10)).
107-
Map(func(_ interface{}) (interface{}, error) {
108-
return 1, nil
109-
}, WithBufferedChannel(11))
110-
assert.Equal(t, 11, cap(obs1.Observe(WithBufferedChannel(13))))
111-
112-
obs2 := obs1.Map(func(_ interface{}) (interface{}, error) {
113-
return 1, nil
114-
}, WithBufferedChannel(12))
115-
assert.Equal(t, 12, cap(obs2.Observe(WithBufferedChannel(13))))
116-
}
117-
118-
func Test_FromFuncs(t *testing.T) {
119-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
83+
func Test_Defer(t *testing.T) {
84+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
12085
next <- Of(1)
12186
next <- Of(2)
12287
next <- Of(3)
@@ -125,8 +90,8 @@ func Test_FromFuncs(t *testing.T) {
12590
Assert(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
12691
}
12792

128-
func Test_FromFuncs_Multiple(t *testing.T) {
129-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
93+
func Test_Defer_Multiple(t *testing.T) {
94+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
13095
next <- Of(1)
13196
next <- Of(2)
13297
done()
@@ -138,8 +103,8 @@ func Test_FromFuncs_Multiple(t *testing.T) {
138103
Assert(context.Background(), t, obs, HasItemsNoParticularOrder(1, 2, 10, 20), HasNotRaisedError())
139104
}
140105

141-
func Test_FromFuncs_Close(t *testing.T) {
142-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
106+
func Test_Defer_Close(t *testing.T) {
107+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
143108
next <- Of(1)
144109
next <- Of(2)
145110
next <- Of(3)
@@ -148,8 +113,8 @@ func Test_FromFuncs_Close(t *testing.T) {
148113
Assert(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
149114
}
150115

151-
func Test_FromFuncs_SingleDup(t *testing.T) {
152-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
116+
func Test_Defer_SingleDup(t *testing.T) {
117+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
153118
next <- Of(1)
154119
next <- Of(2)
155120
next <- Of(3)
@@ -159,8 +124,8 @@ func Test_FromFuncs_SingleDup(t *testing.T) {
159124
Assert(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
160125
}
161126

162-
func Test_FromFuncs_ComposedDup(t *testing.T) {
163-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
127+
func Test_Defer_ComposedDup(t *testing.T) {
128+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
164129
next <- Of(1)
165130
next <- Of(2)
166131
next <- Of(3)
@@ -174,8 +139,8 @@ func Test_FromFuncs_ComposedDup(t *testing.T) {
174139
Assert(context.Background(), t, obs, HasItems(3, 4, 5), HasNotRaisedError())
175140
}
176141

177-
func Test_FromFuncs_ComposedDup_EagerObservation(t *testing.T) {
178-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
142+
func Test_Defer_ComposedDup_EagerObservation(t *testing.T) {
143+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
179144
next <- Of(1)
180145
next <- Of(2)
181146
next <- Of(3)
@@ -186,13 +151,13 @@ func Test_FromFuncs_ComposedDup_EagerObservation(t *testing.T) {
186151
return i.(int) + 1, nil
187152
})
188153
Assert(context.Background(), t, obs, HasItems(3, 4, 5), HasNotRaisedError())
189-
// In the case of an eager observation, we already consumed the items produced by FromFuncs
154+
// In the case of an eager observation, we already consumed the items produced by Defer
190155
// So if we create another subscription, it will be empty
191156
Assert(context.Background(), t, obs, HasNoItem(), HasNotRaisedError())
192157
}
193158

194-
func Test_FromFuncs_Error(t *testing.T) {
195-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
159+
func Test_Defer_Error(t *testing.T) {
160+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
196161
next <- Of(1)
197162
next <- Of(2)
198163
next <- Error(errFoo)
@@ -201,15 +166,15 @@ func Test_FromFuncs_Error(t *testing.T) {
201166
Assert(context.Background(), t, obs, HasItems(1, 2), HasRaisedError(errFoo))
202167
}
203168

204-
func Test_FromFuncs_SimpleCapacity(t *testing.T) {
205-
ch := FromFuncs(func(_ context.Context, _ chan<- Item, done func()) {
169+
func Test_Defer_SimpleCapacity(t *testing.T) {
170+
ch := Defer(func(_ context.Context, _ chan<- Item, done func()) {
206171
done()
207172
}).Observe(WithBufferedChannel(5))
208173
assert.Equal(t, 5, cap(ch))
209174
}
210175

211-
func Test_FromFuncs_ComposedCapacity(t *testing.T) {
212-
obs1 := FromFuncs(func(_ context.Context, _ chan<- Item, done func()) {
176+
func Test_Defer_ComposedCapacity(t *testing.T) {
177+
obs1 := Defer(func(_ context.Context, _ chan<- Item, done func()) {
213178
done()
214179
}).Map(func(_ interface{}) (interface{}, error) {
215180
return 1, nil
@@ -222,6 +187,41 @@ func Test_FromFuncs_ComposedCapacity(t *testing.T) {
222187
assert.Equal(t, 12, cap(obs2.Observe(WithBufferedChannel(13))))
223188
}
224189

190+
func Test_Empty(t *testing.T) {
191+
obs := Empty()
192+
Assert(context.Background(), t, obs, HasNoItems())
193+
}
194+
195+
func Test_FromChannel(t *testing.T) {
196+
ch := make(chan Item)
197+
go func() {
198+
ch <- Of(1)
199+
ch <- Of(2)
200+
ch <- Of(3)
201+
close(ch)
202+
}()
203+
obs := FromChannel(ch)
204+
Assert(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
205+
}
206+
207+
func Test_FromChannel_SimpleCapacity(t *testing.T) {
208+
ch := FromChannel(make(chan Item, 10)).Observe(WithBufferedChannel(11))
209+
assert.Equal(t, 10, cap(ch))
210+
}
211+
212+
func Test_FromChannel_ComposedCapacity(t *testing.T) {
213+
obs1 := FromChannel(make(chan Item, 10)).
214+
Map(func(_ interface{}) (interface{}, error) {
215+
return 1, nil
216+
}, WithBufferedChannel(11))
217+
assert.Equal(t, 11, cap(obs1.Observe(WithBufferedChannel(13))))
218+
219+
obs2 := obs1.Map(func(_ interface{}) (interface{}, error) {
220+
return 1, nil
221+
}, WithBufferedChannel(12))
222+
assert.Equal(t, 12, cap(obs2.Observe(WithBufferedChannel(13))))
223+
}
224+
225225
func Test_FromItem(t *testing.T) {
226226
single := JustItem(Of(1))
227227
Assert(context.Background(), t, single, HasItem(1), HasNotRaisedError())

‎observable_operator_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -593,7 +593,7 @@ func Test_Observable_ReturnError(t *testing.T) {
593593

594594
func Test_Observable_Retry(t *testing.T) {
595595
i := 0
596-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
596+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
597597
next <- Of(1)
598598
next <- Of(2)
599599
if i == 2 {
@@ -609,7 +609,7 @@ func Test_Observable_Retry(t *testing.T) {
609609
}
610610

611611
func Test_Observable_Retry_Error(t *testing.T) {
612-
obs := FromFuncs(func(ctx context.Context, next chan<- Item, done func()) {
612+
obs := Defer(func(ctx context.Context, next chan<- Item, done func()) {
613613
next <- Of(1)
614614
next <- Of(2)
615615
next <- Error(errFoo)

0 commit comments

Comments
 (0)