Performance optimization-- cache the encoding step

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() {
+		buf := new(bytes.Buffer)
+		enc := NewEncoder(buf, false)
+		enc.Encode(evt.wrapped)
+		evt.encoded = buf.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()
+}

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)
+	}
+}

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()
 	}

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),
 	}
 }