From cedc8668e250abbff9b62d91432a1d43c33b424a Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Fri, 4 Nov 2016 14:22:30 -0700 Subject: [PATCH 1/2] Cache encoding optimization for message publication --- cached_encoding_event.go | 43 ++++++++++++++++++++++++++++++++++++++++ encoder.go | 26 +++++++++++++++--------- server.go | 2 +- 3 files changed, 61 insertions(+), 10 deletions(-) create mode 100644 cached_encoding_event.go diff --git a/cached_encoding_event.go b/cached_encoding_event.go new file mode 100644 index 0000000..fdea9e0 --- /dev/null +++ b/cached_encoding_event.go @@ -0,0 +1,43 @@ +package eventsource + +import ( + "bytes" + "sync" +) + +type cachedEncodingEvent struct { + wrapped Event + once *sync.Once + encoded []byte +} + +func makeCachedEncodingEvent(evt Event) cachedEncodingEvent { + return cachedEncodingEvent{ + wrapped: evt, + once: new(sync.Once), + encoded: nil, + } +} + +func (evt cachedEncodingEvent) Encode() []byte { + evt.once.Do(func() { + var buff bytes.Buffer + enc := NewEncoder(&buff, false) + enc.Encode(evt) + evt.encoded = buff.Bytes() + }) + + return evt.encoded +} + +func (evt cachedEncodingEvent) Id() string { + return evt.wrapped.Id() +} + +func (evt cachedEncodingEvent) Event() string { + return evt.wrapped.Event() +} + +func (evt cachedEncodingEvent) Data() string { + return evt.wrapped.Data() +} diff --git a/encoder.go b/encoder.go index e9c7c31..1156152 100644 --- a/encoder.go +++ b/encoder.go @@ -38,19 +38,27 @@ func NewEncoder(w io.Writer, compressed bool) *Encoder { // Encode writes an event in the format specified by the // server-sent events protocol. func (enc *Encoder) Encode(ev Event) error { - for _, field := range encFields { - prefix, value := field.prefix, field.value(ev) - if len(value) == 0 { - continue + if cev, ok := ev.(cachedEncodingEvent); ok { + data := cev.Encode() + if _, err := enc.w.Write(data); err != nil { + return fmt.Errorf("eventsource encode: %v", err) + } + } else { + + for _, field := range encFields { + prefix, value := field.prefix, field.value(ev) + if len(value) == 0 { + continue + } + value = strings.Replace(value, "\n", "\n"+prefix, -1) + if _, err := io.WriteString(enc.w, prefix+value+"\n"); err != nil { + return fmt.Errorf("eventsource encode: %v", err) + } } - value = strings.Replace(value, "\n", "\n"+prefix, -1) - if _, err := io.WriteString(enc.w, prefix+value+"\n"); err != nil { + if _, err := io.WriteString(enc.w, "\n"); err != nil { return fmt.Errorf("eventsource encode: %v", err) } } - if _, err := io.WriteString(enc.w, "\n"); err != nil { - return fmt.Errorf("eventsource encode: %v", err) - } if enc.compressed { return enc.w.(*gzip.Writer).Flush() } diff --git a/server.go b/server.go index 5090be0..9c66467 100644 --- a/server.go +++ b/server.go @@ -113,7 +113,7 @@ func (srv *Server) Register(channel string, repo Repository) { func (srv *Server) Publish(channels []string, ev Event) { srv.pub <- &outbound{ channels: channels, - event: ev, + event: makeCachedEncodingEvent(ev), } } From 57587d19a30c38aa586abbcab4ffa4df865d498d Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Fri, 4 Nov 2016 14:51:17 -0700 Subject: [PATCH 2/2] Improvements and benchmarks for cached encoding events --- cached_encoding_event.go | 8 ++++---- codec_test.go | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/cached_encoding_event.go b/cached_encoding_event.go index fdea9e0..8c74e41 100644 --- a/cached_encoding_event.go +++ b/cached_encoding_event.go @@ -21,10 +21,10 @@ func makeCachedEncodingEvent(evt Event) cachedEncodingEvent { func (evt cachedEncodingEvent) Encode() []byte { evt.once.Do(func() { - var buff bytes.Buffer - enc := NewEncoder(&buff, false) - enc.Encode(evt) - evt.encoded = buff.Bytes() + buf := new(bytes.Buffer) + enc := NewEncoder(buf, false) + enc.Encode(evt.wrapped) + evt.encoded = buf.Bytes() }) return evt.encoded diff --git a/codec_test.go b/codec_test.go index c84a871..a6c8936 100644 --- a/codec_test.go +++ b/codec_test.go @@ -42,3 +42,43 @@ func TestRoundTrip(t *testing.T) { } } } + +func TestCachedRoundTrip(t *testing.T) { + buf := new(bytes.Buffer) + enc := NewEncoder(buf, false) + dec := NewDecoder(buf) + for _, tt := range encoderTests { + want := tt.event + if err := enc.Encode(makeCachedEncodingEvent(want)); err != nil { + t.Fatal(err) + } + if buf.String() != tt.output { + t.Errorf("Expected: %s Got: %s", tt.output, buf.String()) + } + ev, err := dec.Decode() + if err != nil { + t.Fatal(err) + } + if ev.Id() != want.Id() || ev.Event() != want.Event() || ev.Data() != want.Data() { + t.Errorf("Expected: %s %s %s Got: %s %s %s", want.Id(), want.Event(), want.Data(), ev.Id(), ev.Event(), ev.Data()) + } + } +} + +func BenchmarkEncoding(b *testing.B) { + event := encoderTests[0].event + for n := 0; n < b.N; n++ { + buf := new(bytes.Buffer) + enc := NewEncoder(buf, false) + enc.Encode(event) + } +} + +func BenchmarkCachedEncoding(b *testing.B) { + event := makeCachedEncodingEvent(encoderTests[0].event) + for n := 0; n < b.N; n++ { + buf := new(bytes.Buffer) + enc := NewEncoder(buf, false) + enc.Encode(event) + } +}