Skip to content

Commit d7194bc

Browse files
fraironfrairon
committed
Add support for pooling in codecs, partition_table, view etc.
1 parent 2f90eeb commit d7194bc

File tree

31 files changed

+494
-326
lines changed

31 files changed

+494
-326
lines changed

‎codec.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,17 @@
11
package goka
22

3+
import "io"
4+
35
// Codec decodes and encodes from and to []byte
46
type Codec interface {
57
Encode(value interface{}) (data []byte, err error)
68
Decode(data []byte) (value interface{}, err error)
9+
DecodeP(data []byte) (value interface{}, closer io.Closer, err error)
10+
}
11+
12+
// FuncCloser implements io.Closer-interface for convenience
13+
type FuncCloser func() error
14+
15+
func (f FuncCloser) Close() error {
16+
return f()
717
}

‎codec/closer.go‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package codec
2+
3+
type Closer interface {
4+
Close()
5+
}
6+
7+
type nullCloser struct{}
8+
9+
func (n *nullCloser) Close() error { return nil }
10+
11+
var NoopCloser = new(nullCloser)

‎codec/codec.go‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package codec
22

33
import (
44
"fmt"
5+
"io"
56
"strconv"
67
)
78

@@ -23,6 +24,11 @@ func (d *Bytes) Decode(data []byte) (interface{}, error) {
2324
return data, nil
2425
}
2526

27+
// Decode of defaultCodec simply returns the data
28+
func (d *Bytes) DecodeP(data []byte) (interface{}, io.Closer, error) {
29+
return data, NoopCloser, nil
30+
}
31+
2632
// String is a commonly used codec to encode and decode string <-> []byte
2733
type String struct{}
2834

@@ -40,6 +46,10 @@ func (c *String) Decode(data []byte) (interface{}, error) {
4046
return string(data), nil
4147
}
4248

49+
func (c *String) DecodeP(data []byte) (interface{}, io.Closer, error) {
50+
return string(data), NoopCloser, nil
51+
}
52+
4353
// Int64 is a commonly used codec to encode and decode string <-> []byte
4454
type Int64 struct{}
4555

@@ -60,3 +70,8 @@ func (c *Int64) Decode(data []byte) (interface{}, error) {
6070
}
6171
return intVal, nil
6272
}
73+
74+
func (c *Int64) DecodeP(data []byte) (interface{}, io.Closer, error) {
75+
dec, err := c.Decode(data)
76+
return dec, NoopCloser, err
77+
}

‎context.go‎

Lines changed: 40 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,8 @@ type cbContext struct {
170170
// tracking statistics for the output topic
171171
trackOutputStats func(ctx context.Context, topic string, size int)
172172

173+
deferreds []func() error
174+
173175
msg *message
174176
done bool
175177
counters struct {
@@ -310,17 +312,25 @@ func (ctx *cbContext) Join(topic Table) interface{} {
310312
if !ok {
311313
ctx.Fail(fmt.Errorf("table %s not subscribed", topic))
312314
}
313-
data, err := v.st.Get(ctx.Key())
315+
data, getCloser, err := v.st.GetP(ctx.Key())
314316
if err != nil {
315317
ctx.Fail(fmt.Errorf("error getting key %s of table %s: %v", ctx.Key(), topic, err))
316318
} else if data == nil {
317319
return nil
318320
}
319321

320-
value, err := ctx.graph.codec(string(topic)).Decode(data)
322+
if getCloser != nil {
323+
ctx.addDeferred(getCloser.Close)
324+
}
325+
326+
value, decodeCloser, err := ctx.graph.codec(string(topic)).DecodeP(data)
321327
if err != nil {
322328
ctx.Fail(fmt.Errorf("error decoding value key %s of table %s: %v", ctx.Key(), topic, err))
323329
}
330+
if decodeCloser != nil {
331+
ctx.addDeferred(decodeCloser.Close)
332+
}
333+
324334
return value
325335
}
326336

@@ -332,10 +342,13 @@ func (ctx *cbContext) Lookup(topic Table, key string) interface{} {
332342
if !ok {
333343
ctx.Fail(fmt.Errorf("topic %s not subscribed", topic))
334344
}
335-
val, err := v.Get(key)
345+
val, getCloser, err := v.GetP(key)
336346
if err != nil {
337347
ctx.Fail(fmt.Errorf("error getting key %s of table %s: %v", key, topic, err))
338348
}
349+
if getCloser != nil {
350+
ctx.addDeferred(getCloser.Close)
351+
}
339352
return val
340353
}
341354

@@ -345,17 +358,26 @@ func (ctx *cbContext) valueForKey(key string) (interface{}, error) {
345358
return nil, fmt.Errorf("Cannot access state in stateless processor")
346359
}
347360

348-
data, err := ctx.table.Get(key)
361+
data, closer, err := ctx.table.Get(key)
349362
if err != nil {
350363
return nil, fmt.Errorf("error reading value: %v", err)
351-
} else if data == nil {
364+
}
365+
if closer != nil {
366+
ctx.addDeferred(closer.Close)
367+
}
368+
369+
if data == nil {
352370
return nil, nil
353371
}
354372

355-
value, err := ctx.graph.GroupTable().Codec().Decode(data)
373+
value, decodeCloser, err := ctx.graph.GroupTable().Codec().DecodeP(data)
356374
if err != nil {
357375
return nil, fmt.Errorf("error decoding value: %v", err)
358376
}
377+
if decodeCloser != nil {
378+
ctx.addDeferred(decodeCloser.Close)
379+
}
380+
359381
return value, nil
360382
}
361383

@@ -450,6 +472,14 @@ func (ctx *cbContext) tryCommit(err error) {
450472
if ctx.errors.ErrorOrNil() != nil {
451473
ctx.asyncFailer(fmt.Errorf("could not commit message with key '%s': %w", ctx.Key(), ctx.errors.ErrorOrNil()))
452474
} else {
475+
476+
// execute deferred commit functions in reverse order
477+
for i := len(ctx.deferreds) - 1; i >= 0; i-- {
478+
if err := ctx.deferreds[i](); err != nil {
479+
ctx.asyncFailer(fmt.Errorf("error executing context deferred: %w", err))
480+
}
481+
}
482+
453483
ctx.commit()
454484
}
455485

@@ -483,3 +513,7 @@ func (ctx *cbContext) DeferCommit() func(err error) {
483513
})
484514
}
485515
}
516+
517+
func (ctx *cbContext) addDeferred(def func() error) {
518+
ctx.deferreds = append(ctx.deferreds, def)
519+
}

‎context_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -380,9 +380,9 @@ func TestContext_GetSetStateful(t *testing.T) {
380380
}
381381
)
382382

383-
st.EXPECT().Get(key).Return(nil, nil)
383+
st.EXPECT().GetP(key).Return(nil, nil, nil)
384384
st.EXPECT().Set(key, []byte(value)).Return(nil)
385-
st.EXPECT().Get(key).Return([]byte(value), nil)
385+
st.EXPECT().GetP(key).Return([]byte(value), codec.NoopCloser, nil)
386386

387387
graph := DefineGroup(group, Persist(new(codec.String)))
388388
ctx := &cbContext{
@@ -537,11 +537,11 @@ func TestContext_Join(t *testing.T) {
537537
syncFailer: func(err error) { panic(err) },
538538
}
539539

540-
st.EXPECT().Get(key).Return([]byte(value), nil)
540+
st.EXPECT().GetP(key).Return([]byte(value), codec.NoopCloser, nil)
541541
v := ctx.Join(table)
542542
require.Equal(t, value, v)
543543

544-
st.EXPECT().Get(key).Return(nil, errSome)
544+
st.EXPECT().GetP(key).Return(nil, codec.NoopCloser, errSome)
545545
require.Panics(t, func() { ctx.Join(table) })
546546

547547
require.Panics(t, func() { ctx.Join("other-table") })
@@ -586,11 +586,11 @@ func TestContext_Lookup(t *testing.T) {
586586
syncFailer: func(err error) { panic(err) },
587587
}
588588

589-
st.EXPECT().Get(key).Return([]byte(value), nil)
589+
st.EXPECT().GetP(key).Return([]byte(value), codec.NoopCloser, nil)
590590
v := ctx.Lookup(table, key)
591591
require.Equal(t, value, v)
592592

593-
st.EXPECT().Get(key).Return(nil, errSome)
593+
st.EXPECT().GetP(key).Return(nil, codec.NoopCloser, errSome)
594594
require.Panics(t, func() { ctx.Lookup(table, key) })
595595
require.Panics(t, func() { ctx.Lookup("other-table", key) })
596596

‎examples/2-clicks/main.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"encoding/json"
66
"fmt"
7+
"io"
78
"log"
89
"net/http"
910
"time"
@@ -58,6 +59,11 @@ func (jc *userCodec) Decode(data []byte) (interface{}, error) {
5859
return &c, nil
5960
}
6061

62+
func (jc *userCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
63+
dec, err := jc.Decode(data)
64+
return dec, codec.NoopCloser, err
65+
}
66+
6167
func runEmitter() {
6268
emitter, err := goka.NewEmitter(brokers, topic,
6369
new(codec.String))

‎examples/3-messaging/blocker/blocker.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,10 @@ package blocker
33
import (
44
"context"
55
"encoding/json"
6+
"io"
7+
68
"github.com/lovoo/goka"
9+
"github.com/lovoo/goka/codec"
710
"github.com/lovoo/goka/examples/3-messaging/topicinit"
811
)
912

@@ -28,6 +31,11 @@ func (c *BlockEventCodec) Decode(data []byte) (interface{}, error) {
2831
return &m, json.Unmarshal(data, &m)
2932
}
3033

34+
func (c *BlockEventCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
35+
dec, err := c.Decode(data)
36+
return dec, codec.NoopCloser, err
37+
}
38+
3139
type BlockValue struct {
3240
Blocked bool
3341
}
@@ -42,6 +50,11 @@ func (c *BlockValueCodec) Decode(data []byte) (interface{}, error) {
4250
return &m, json.Unmarshal(data, &m)
4351
}
4452

53+
func (c *BlockValueCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
54+
dec, err := c.Decode(data)
55+
return dec, codec.NoopCloser, err
56+
}
57+
4558
func block(ctx goka.Context, msg interface{}) {
4659
var s *BlockValue
4760
if v := ctx.Value(); v == nil {

‎examples/3-messaging/collector/collector.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,11 @@ package collector
33
import (
44
"context"
55
"encoding/json"
6+
"io"
7+
68
"github.com/lovoo/goka"
7-
"github.com/lovoo/goka/examples/3-messaging"
9+
"github.com/lovoo/goka/codec"
10+
messaging "github.com/lovoo/goka/examples/3-messaging"
811
"github.com/lovoo/goka/examples/3-messaging/topicinit"
912
)
1013

@@ -27,6 +30,11 @@ func (c *MessageListCodec) Decode(data []byte) (interface{}, error) {
2730
return m, err
2831
}
2932

33+
func (c *MessageListCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
34+
dec, err := c.Decode(data)
35+
return dec, codec.NoopCloser, err
36+
}
37+
3038
func collect(ctx goka.Context, msg interface{}) {
3139
var ml []messaging.Message
3240
if v := ctx.Value(); v != nil {

‎examples/3-messaging/detector/detector.go‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,11 @@ package detector
33
import (
44
"context"
55
"encoding/json"
6+
"io"
7+
68
"github.com/lovoo/goka"
7-
"github.com/lovoo/goka/examples/3-messaging"
9+
"github.com/lovoo/goka/codec"
10+
messaging "github.com/lovoo/goka/examples/3-messaging"
811
"github.com/lovoo/goka/examples/3-messaging/blocker"
912
"github.com/lovoo/goka/examples/3-messaging/topicinit"
1013
)
@@ -14,9 +17,7 @@ const (
1417
maxRate = 0.5
1518
)
1619

17-
var (
18-
group goka.Group = "detector"
19-
)
20+
var group goka.Group = "detector"
2021

2122
type Counters struct {
2223
Sent int
@@ -34,6 +35,11 @@ func (c *CountersCodec) Decode(data []byte) (interface{}, error) {
3435
return &m, json.Unmarshal(data, &m)
3536
}
3637

38+
func (c *CountersCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
39+
dec, err := c.Decode(data)
40+
return dec, codec.NoopCloser, err
41+
}
42+
3743
func getValue(ctx goka.Context) *Counters {
3844
if v := ctx.Value(); v != nil {
3945
return v.(*Counters)

‎examples/3-messaging/message.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,10 @@ package messaging
22

33
import (
44
"encoding/json"
5+
"io"
56

67
"github.com/lovoo/goka"
8+
"github.com/lovoo/goka/codec"
79
)
810

911
var (
@@ -28,6 +30,11 @@ func (c *MessageCodec) Decode(data []byte) (interface{}, error) {
2830
return &m, json.Unmarshal(data, &m)
2931
}
3032

33+
func (c *MessageCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
34+
dec, err := c.Decode(data)
35+
return dec, codec.NoopCloser, err
36+
}
37+
3138
type MessageListCodec struct{}
3239

3340
func (c *MessageListCodec) Encode(value interface{}) ([]byte, error) {
@@ -39,3 +46,8 @@ func (c *MessageListCodec) Decode(data []byte) (interface{}, error) {
3946
err := json.Unmarshal(data, &m)
4047
return m, err
4148
}
49+
50+
func (c *MessageListCodec) DecodeP(data []byte) (interface{}, io.Closer, error) {
51+
dec, err := c.Decode(data)
52+
return dec, codec.NoopCloser, err
53+
}

0 commit comments

Comments
 (0)