+
+{{ end }}
+{{ define "footer"}} {{ partial "footer" .}} {{end}}
diff --git a/docs/layouts/partials/footer.html b/docs/layouts/partials/footer.html
new file mode 100644
index 000000000..80ae1b269
--- /dev/null
+++ b/docs/layouts/partials/footer.html
@@ -0,0 +1,14 @@
+
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/collapse-box.html b/docs/layouts/shortcodes/collapse-box.html
new file mode 100644
index 000000000..ed7e6898e
--- /dev/null
+++ b/docs/layouts/shortcodes/collapse-box.html
@@ -0,0 +1,3 @@
+
+ {{ .Inner }}
+
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/collapse-toggle.html b/docs/layouts/shortcodes/collapse-toggle.html
new file mode 100644
index 000000000..d541b5490
--- /dev/null
+++ b/docs/layouts/shortcodes/collapse-toggle.html
@@ -0,0 +1 @@
+
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/collapse.html b/docs/layouts/shortcodes/collapse.html
new file mode 100644
index 000000000..95d838c70
--- /dev/null
+++ b/docs/layouts/shortcodes/collapse.html
@@ -0,0 +1,3 @@
+
+ {{ .Inner }}
+
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/contact-email.html b/docs/layouts/shortcodes/contact-email.html
new file mode 100644
index 000000000..7e80368d6
--- /dev/null
+++ b/docs/layouts/shortcodes/contact-email.html
@@ -0,0 +1,3 @@
+{{/* hack: we cannot use variables directly on the markdown, only shortcodes */}}
+
+{{ $.Page.Site.Params.Email }}
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/highlight.html b/docs/layouts/shortcodes/highlight.html
new file mode 100644
index 000000000..8df83db38
--- /dev/null
+++ b/docs/layouts/shortcodes/highlight.html
@@ -0,0 +1 @@
+{{ highlight .Inner (.Get "language" | default "go") "" }}
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/load-snippet-partial.html b/docs/layouts/shortcodes/load-snippet-partial.html
new file mode 100644
index 000000000..8fa952c88
--- /dev/null
+++ b/docs/layouts/shortcodes/load-snippet-partial.html
@@ -0,0 +1,67 @@
+{{ $file := (.Get "file") }}
+{{ $content := readFile $file }}
+
+{{ $first_line_contains := (.Get "first_line_contains") }}
+{{ $last_line_contains := (.Get "last_line_contains") }}
+
+{{ $show_line := false }}
+
+{{/*if true, first or last line was not found*/}}
+{{ $first_line_found := false}}
+{{ $last_line_found := false}}
+
+{{ $padding_after := (.Get "padding_after" | default "0" | int) }}
+
+{{ $first_line_num := 0 }}
+{{ $last_line_num := 0 }}
+
+{{ $linkFile := $file }}
+{{ if in $file "content/src-link/" }}
+ {{ $linkFile = replace $linkFile "content/src-link/" "" }}
+{{ end }}
+
+{{ $lines := slice }}
+
+{{ range $elem_key, $elem_val := split $content "\n" }}
+ {{ $line_num := (add $elem_key 1) }}
+
+ {{ if and (not $first_line_found) (in $elem_val $first_line_contains) }}
+ {{ if ne $elem_key 0 }}
+ {{ $lines = $lines | append "// ..." }}
+ {{ end }}
+
+ {{ $show_line = true }}
+ {{ $first_line_found = true}}
+ {{ $first_line_num = $line_num }}
+ {{ end }}
+
+ {{ if $show_line }}
+ {{ $lines = $lines | append $elem_val }}
+ {{ end }}
+
+ {{ if and ($first_line_found) (in $elem_val $last_line_contains) (ne $last_line_contains "") }}
+ {{ $last_line_found = true }}
+ {{ end }}
+
+ {{ if and $last_line_found $show_line }}
+ {{ if gt $padding_after 0 }}
+ {{ $padding_after = sub $padding_after 1}}
+ {{ else }}
+ {{ $lines = $lines | append "// ..." }}
+ {{ $show_line = false }}
+ {{ $last_line_num = $line_num }}
+ {{ end }}
+ {{ end }}
+{{ end }}
+
+Full source: [{{ $linkFile }}](https://github.com/ThreeDotsLabs/watermill/tree/master/{{ $linkFile }}{{ if ne $first_line_num 0 }}#L{{ $first_line_num }}{{ end }})
+
+{{ highlight (delimit $lines "\n") (.Get "type" | default "go") "" }}
+
+{{if not $first_line_found }}
+ {{ errorf "`first_line_contains` %s not found in %s snippet" $first_line_contains $file }}
+{{end}}
+
+{{if and (not $last_line_found) (ne $last_line_contains "") }}
+ {{ errorf "`last_line_contains` %s not found in %s snippet" $last_line_contains $file }}
+{{end}}
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/load-snippet.html b/docs/layouts/shortcodes/load-snippet.html
new file mode 100644
index 000000000..032f980c6
--- /dev/null
+++ b/docs/layouts/shortcodes/load-snippet.html
@@ -0,0 +1,26 @@
+{{ $file := (.Get "file") }}
+{{ $content := readFile $file }}
+
+{{ $start_line := (.Get "start_line") | default "0" }}
+{{ $end_line := (.Get "end_line") | default "0" }}
+
+{{ $has_start_line := (ne $start_line "0") }}
+{{ $has_end_line := (ne $end_line "0") }}
+
+{{ $lines := slice }}
+
+{{ $linkFile := $file }}
+{{ if in $file "content/src-link/" }}
+ {{ $linkFile = replace $linkFile "content/src-link/" "" }}
+{{ end }}
+
+Full source: [{{ $linkFile }}](https://github.com/ThreeDotsLabs/watermill/tree/master/{{ $linkFile }})
+
+{{ range $elem_key, $elem_val := split $content "\n" }}
+ {{if and (or (not $has_start_line) (ge (add $elem_key 1) ($start_line | int))) (or (not $has_end_line) (le (add $elem_key 1) ($end_line | int)))}}
+ {{ $lines = $lines | append $elem_val }}
+ {{ end }}
+{{ end }}
+
+
+{{ highlight (delimit $lines "\n") (.Get "type" | default "go") "" }}
diff --git a/docs/layouts/shortcodes/message.html b/docs/layouts/shortcodes/message.html
new file mode 100644
index 000000000..16fc58452
--- /dev/null
+++ b/docs/layouts/shortcodes/message.html
@@ -0,0 +1,4 @@
+
+ {{ if ne (.Get "title") "" }}
{{ .Get "title" }}
{{ end }}
+ {{ .Inner }}
+
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/render-md.html b/docs/layouts/shortcodes/render-md.html
new file mode 100644
index 000000000..825c3f939
--- /dev/null
+++ b/docs/layouts/shortcodes/render-md.html
@@ -0,0 +1,11 @@
+{{/*
+hugo pls, why i just cannot render shortcode as markdown,
+used for:
+
+{{% render-md %}}
+{{% load-snippet-partial (...) %}}
+{{% /render-md %}}
+
+bacause it is rendered as raw text by default
+*/}}
+{{.Inner}}
\ No newline at end of file
diff --git a/docs/layouts/shortcodes/tabs-tab.html b/docs/layouts/shortcodes/tabs-tab.html
new file mode 100644
index 000000000..00885f920
--- /dev/null
+++ b/docs/layouts/shortcodes/tabs-tab.html
@@ -0,0 +1,3 @@
+
+ {{ .Inner }}
+
diff --git a/docs/layouts/shortcodes/tabs.html b/docs/layouts/shortcodes/tabs.html
new file mode 100644
index 000000000..657af6448
--- /dev/null
+++ b/docs/layouts/shortcodes/tabs.html
@@ -0,0 +1,13 @@
+
+
+ {{.Inner}}
+
\ No newline at end of file
diff --git a/docs/static/css/custom.css b/docs/static/css/custom.css
new file mode 100644
index 000000000..b9dd3a09c
--- /dev/null
+++ b/docs/static/css/custom.css
@@ -0,0 +1,24 @@
+pre, pre code {
+ background: #f8f8f8 !important;
+}
+
+#kube-features .row:first-child {
+ border-bottom: none;
+}
+
+.message > p {
+ margin-bottom: 0px;
+}
+
+/* fix for stupid css */
+.my-collapse div {
+ border: none;
+ padding: 0;
+ margin-bottom: 0;
+}
+
+.my-collapse > div.collapse-box {
+ border: 1px solid rgba(0, 0, 0, 0.1);
+ padding: 24px 32px 1px;
+ margin-bottom: 1px;
+}
\ No newline at end of file
diff --git a/docs/static/img/watermill-router.svg b/docs/static/img/watermill-router.svg
new file mode 100644
index 000000000..5b174c939
--- /dev/null
+++ b/docs/static/img/watermill-router.svg
@@ -0,0 +1 @@
+
\ No newline at end of file
diff --git a/log.go b/log.go
index 2630d9919..74ed4db2e 100644
--- a/log.go
+++ b/log.go
@@ -45,7 +45,7 @@ type StdLoggerAdapter struct {
func NewStdLogger(debug, trace bool) LoggerAdapter {
l := log.New(os.Stderr, "[watermill] ", log.LstdFlags|log.Lmicroseconds|log.Lshortfile)
- a := &StdLoggerAdapter{InfoLogger: l}
+ a := &StdLoggerAdapter{InfoLogger: l, ErrorLogger: l}
if debug {
a.DebugLogger = l
@@ -58,7 +58,7 @@ func NewStdLogger(debug, trace bool) LoggerAdapter {
}
func (l *StdLoggerAdapter) Error(msg string, err error, fields LogFields) {
- l.log(l.TraceLogger, "ERROR", msg, fields.Add(LogFields{"err": err}))
+ l.log(l.ErrorLogger, "ERROR", msg, fields.Add(LogFields{"err": err}))
}
func (l *StdLoggerAdapter) Info(msg string, fields LogFields) {
diff --git a/message/infrastructure/gochannel/pubsub.go b/message/infrastructure/gochannel/pubsub.go
index 8b778dbfb..14ab2d7a8 100644
--- a/message/infrastructure/gochannel/pubsub.go
+++ b/message/infrastructure/gochannel/pubsub.go
@@ -16,7 +16,12 @@ type subscriber struct {
outputChannel chan *message.Message
}
-type goChannel struct {
+// GoChannel is the simplest Pub/Sub implementation.
+// It is based on Golang's channels which are sent within the process.
+//
+// GoChannel has no global state,
+// that means that you need to use the same instance for Publishing and Subscribing!
+type GoChannel struct {
sendTimeout time.Duration
buffer int64
@@ -29,7 +34,7 @@ type goChannel struct {
}
func NewGoChannel(buffer int64, logger watermill.LoggerAdapter, sendTimeout time.Duration) message.PubSub {
- return &goChannel{
+ return &GoChannel{
sendTimeout: sendTimeout,
buffer: buffer,
@@ -39,7 +44,11 @@ func NewGoChannel(buffer int64, logger watermill.LoggerAdapter, sendTimeout time
}
}
-func (g *goChannel) Publish(topic string, messages ...*message.Message) error {
+// Publish in GoChannel is blocking until all consumers consume and acknowledge the message.
+// Sending message to one subscriber has timeout equal to GoChannel.sendTimeout configured via constructor.
+//
+// Messages are not persisted. If there are no subscribers and message is produced it will be gone.
+func (g *GoChannel) Publish(topic string, messages ...*message.Message) error {
for _, msg := range messages {
if err := g.sendMessage(topic, msg); err != nil {
return err
@@ -49,7 +58,7 @@ func (g *goChannel) Publish(topic string, messages ...*message.Message) error {
return nil
}
-func (g *goChannel) sendMessage(topic string, message *message.Message) error {
+func (g *GoChannel) sendMessage(topic string, message *message.Message) error {
messageLogFields := watermill.LogFields{
"message_uuid": message.UUID,
}
@@ -92,7 +101,11 @@ func (g *goChannel) sendMessage(topic string, message *message.Message) error {
return nil
}
-func (g *goChannel) Subscribe(topic string) (chan *message.Message, error) {
+// Subscribe returns channel to which all published messages are sent.
+// Messages are not persisted. If there are no subscribers and message is produced it will be gone.
+//
+// There are no consumer groups support etc. Every consumer will receive every produced message.
+func (g *GoChannel) Subscribe(topic string) (chan *message.Message, error) {
g.subscribersLock.Lock()
defer g.subscribersLock.Unlock()
@@ -109,7 +122,7 @@ func (g *goChannel) Subscribe(topic string) (chan *message.Message, error) {
return s.outputChannel, nil
}
-func (g *goChannel) Close() error {
+func (g *GoChannel) Close() error {
g.subscribersLock.Lock()
defer g.subscribersLock.Unlock()
diff --git a/message/infrastructure/http/subscriber.go b/message/infrastructure/http/subscriber.go
index fe74ff5fa..10128a6a0 100644
--- a/message/infrastructure/http/subscriber.go
+++ b/message/infrastructure/http/subscriber.go
@@ -11,6 +11,7 @@ import (
type UnmarshalMessageFunc func(topic string, request *http.Request) (*message.Message, error)
+// Subscriber can subscribe to HTTP requests and create Watermill's messages based on them.
type Subscriber struct {
router chi.Router
server *http.Server
@@ -24,6 +25,13 @@ type Subscriber struct {
closed bool
}
+// NewSubscriber creates new Subscriber.
+//
+// addr is TCP address to listen on
+//
+// unmarshalMessageFunc is function which converts HTTP request to Watermill's message.
+//
+// logger is Watermill's logger.
func NewSubscriber(addr string, unmarshalMessageFunc UnmarshalMessageFunc, logger watermill.LoggerAdapter) (*Subscriber, error) {
r := chi.NewRouter()
s := &http.Server{Addr: addr, Handler: r}
@@ -39,17 +47,23 @@ func NewSubscriber(addr string, unmarshalMessageFunc UnmarshalMessageFunc, logge
}, nil
}
-func (s *Subscriber) Subscribe(topic string) (chan *message.Message, error) {
+// Subscribe adds HTTP handler which will listen in provided url for messages.
+//
+// Subscribe needs to be called before `StartHTTPServer`.
+//
+// When request is sent, it will wait for the `Ack`. When Ack is received 200 HTTP status wil be sent.
+// When Nack is sent, 500 HTTP status will be sent.
+func (s *Subscriber) Subscribe(url string) (chan *message.Message, error) {
messages := make(chan *message.Message)
s.outputChannelsLock.Lock()
s.outputChannels = append(s.outputChannels, messages)
s.outputChannelsLock.Unlock()
- baseLogFields := watermill.LogFields{"topic": topic}
+ baseLogFields := watermill.LogFields{"url": url}
- s.router.Post(topic, func(w http.ResponseWriter, r *http.Request) {
- msg, err := s.unmarshalMessageFunc(topic, r)
+ s.router.Post(url, func(w http.ResponseWriter, r *http.Request) {
+ msg, err := s.unmarshalMessageFunc(url, r)
if err != nil {
s.logger.Info("Cannot unmarshal message", baseLogFields.Add(watermill.LogFields{"err": err}))
w.WriteHeader(http.StatusBadRequest)
diff --git a/message/infrastructure/kafka/marshaler.go b/message/infrastructure/kafka/marshaler.go
index 8b87c598b..9eb08ec7a 100644
--- a/message/infrastructure/kafka/marshaler.go
+++ b/message/infrastructure/kafka/marshaler.go
@@ -6,10 +6,14 @@ import (
"github.com/pkg/errors"
)
+const UUIDHeaderKey = "_watermill_message_uuid"
+
+// Marshaler marshals Watermill's message to Kafka message.
type Marshaler interface {
Marshal(topic string, msg *message.Message) (*confluentKafka.Message, error)
}
+// Unmarshaler unmarshals Kafka's message to Watermill's message.
type Unmarshaler interface {
Unmarshal(*confluentKafka.Message) (*message.Message, error)
}
@@ -19,8 +23,6 @@ type MarshalerUnmarshaler interface {
Unmarshaler
}
-const UUIDHeaderKey = "_watermill_message_uuid"
-
type DefaultMarshaler struct{}
func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*confluentKafka.Message, error) {
diff --git a/message/infrastructure/kafka/publisher.go b/message/infrastructure/kafka/publisher.go
index 47af1bc62..5bea9554f 100644
--- a/message/infrastructure/kafka/publisher.go
+++ b/message/infrastructure/kafka/publisher.go
@@ -21,7 +21,7 @@ func NewPublisher(brokers []string, marshaler Marshaler, kafkaConfigOverwrite ka
"bootstrap.servers": strings.Join(brokers, ","),
"queue.buffering.max.messages": 10000000,
"queue.buffering.max.kbytes": 2097151,
- "debug": ",",
+ "debug": ",",
}
if err := mergeConfluentConfigs(config, kafkaConfigOverwrite); err != nil {
@@ -40,6 +40,10 @@ func NewCustomPublisher(producer *kafka.Producer, marshaler Marshaler) (message.
return &confluentPublisher{producer, marshaler, false}, nil
}
+// Publish publishes message to Kafka.
+//
+// Publish is blocking and wait for ack from Kafka.
+// When one of messages delivery fails - function is interrupted.
func (p confluentPublisher) Publish(topic string, msgs ...*message.Message) error {
if p.closed {
return errors.New("publisher closed")
diff --git a/message/infrastructure/kafka/subscriber.go b/message/infrastructure/kafka/subscriber.go
index 0fe0791e7..45f9a3c0a 100644
--- a/message/infrastructure/kafka/subscriber.go
+++ b/message/infrastructure/kafka/subscriber.go
@@ -29,15 +29,25 @@ type confluentSubscriber struct {
}
type SubscriberConfig struct {
+ // Kafka brokers list.
Brokers []string
- ConsumerGroup string
+ // Kafka consumer group.
+ ConsumerGroup string
+ // When we want to consume without consumer group, you should set it to true.
+ // In practice you will receive all messages sent to the topic.
NoConsumerGroup bool
+ // Action to take when there is no initial offset in offset store or the desired offset is out of range.
+ // Available options: smallest, earliest, beginning, largest, latest, end, error.
AutoOffsetReset string
+ // How much consumers should be spawned.
+ // Every consumer will receive messages for their own partition so messages order will be preserved.
ConsumersCount int
+ // Passing librdkafka options.
+ // Available options: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
KafkaConfigOverwrite kafka.ConfigMap
}
@@ -130,6 +140,9 @@ func DefaultConfluentConsumerConstructor(config SubscriberConfig) (*kafka.Consum
return kafka.NewConsumer(kafkaConfig)
}
+// Subscribe subscribers for messages in Kafka.
+//
+// They are multiple subscribers spawned
func (s *confluentSubscriber) Subscribe(topic string) (chan *message.Message, error) {
if s.closed {
return nil, errors.New("subscriber closed")
diff --git a/message/message.go b/message/message.go
index f924a5d23..8b087e6dd 100644
--- a/message/message.go
+++ b/message/message.go
@@ -20,16 +20,28 @@ var (
type Payload []byte
type Message struct {
- UUID string // todo - change to []byte?, change to type
-
+ // UUID is an unique identifier of message.
+ //
+ // It is only used by Watermill for debugging.
+ // UUID can be empty.
+ UUID string
+
+ // Metadata contains the message metadata.
+ //
+ // Can be used to store data which doesn't require unmarshaling entire payload.
+ // It is something similar to HTTP request's headers.
Metadata Metadata
+ // Payload is message's payload.
Payload Payload
- ack chan struct{}
- noAck chan struct{}
- ackMutex sync.Mutex
- ackSent ackType
+ // ack is closed, when acknowledge is received.
+ ack chan struct{}
+ // noACk is closed, when negative acknowledge is received.
+ noAck chan struct{}
+
+ ackMutex sync.Mutex
+ ackSentType ackType
}
func NewMessage(uuid string, payload Payload) *Message {
@@ -50,18 +62,23 @@ const (
nack
)
+// Ack sends message's acknowledgement.
+//
+// Ack is not blocking.
+// Ack is idempotent.
+// Error is returned, if Nack is already sent.
func (m *Message) Ack() error {
m.ackMutex.Lock()
defer m.ackMutex.Unlock()
- if m.ackSent == nack {
+ if m.ackSentType == nack {
return ErrAlreadyNacked
}
- if m.ackSent != noAckSent {
+ if m.ackSentType != noAckSent {
return nil
}
- m.ackSent = ack
+ m.ackSentType = ack
if m.noAck == nil {
m.ack = closedchan
} else {
@@ -71,18 +88,23 @@ func (m *Message) Ack() error {
return nil
}
+// Nack sends message's negative acknowledgement.
+//
+// Nack is not blocking.
+// Nack is idempotent.
+// Error is returned, if Ack is already sent.
func (m *Message) Nack() error {
m.ackMutex.Lock()
defer m.ackMutex.Unlock()
- if m.ackSent == ack {
+ if m.ackSentType == ack {
return ErrAlreadyAcked
}
- if m.ackSent != noAckSent {
+ if m.ackSentType != noAckSent {
return nil
}
- m.ackSent = nack
+ m.ackSentType = nack
if m.noAck == nil {
m.noAck = closedchan
@@ -93,10 +115,28 @@ func (m *Message) Nack() error {
return nil
}
+// Acked returns channel which is closed when acknowledgement is sent.
+//
+// Usage:
+// select {
+// case <-message.Acked():
+// // ack received
+// case <-message.Nacked():
+// // nack received
+// }
func (m *Message) Acked() <-chan struct{} {
return m.ack
}
+// Nacked returns channel which is closed when negative acknowledgement is sent.
+//
+// Usage:
+// select {
+// case <-message.Acked():
+// // ack received
+// case <-message.Nacked():
+// // nack received
+// }
func (m *Message) Nacked() <-chan struct{} {
return m.noAck
}
diff --git a/message/publisher.go b/message/publisher.go
index bf3f40e25..c1f553628 100644
--- a/message/publisher.go
+++ b/message/publisher.go
@@ -1,10 +1,18 @@
package message
type publisher interface {
+ // Publish publishes provided messages to given topic.
+ //
+ // Publish can be synchronous or asynchronous - it depends of implementation.
+ //
+ // Most publishers implementations doesn't support atomic publishing of messages.
+ // That means, that when publishing one of messages failed next messages will be not published.
Publish(topic string, messages ...*Message) error
}
type Publisher interface {
publisher
+
+ // Close should flush unsent messages, if publisher is async.
Close() error
}
diff --git a/message/pubsub.go b/message/pubsub.go
index 17881a346..e9d87c6dc 100644
--- a/message/pubsub.go
+++ b/message/pubsub.go
@@ -9,6 +9,10 @@ type PubSub interface {
Close() error
}
+func NewPubSub(publisher Publisher, subscriber Subscriber) PubSub {
+ return pubSub{publisher, subscriber}
+}
+
type pubSub struct {
Publisher
Subscriber
@@ -32,7 +36,3 @@ func (p pubSub) Close() error {
return errors.New(errMsg)
}
-
-func NewPubSub(publisher Publisher, subscriber Subscriber) PubSub {
- return pubSub{publisher, subscriber}
-}
diff --git a/message/router.go b/message/router.go
index 20fa0b4a1..7670ba0c6 100644
--- a/message/router.go
+++ b/message/router.go
@@ -10,13 +10,40 @@ import (
"github.com/pkg/errors"
)
+// HandlerFunc is function called when message is received.
+//
+// msg.Ack() is called automatically when HandlerFunc doesn't return error.
+// When HandlerFunc returns error, msg.Nack() is called.
+// When msg.Ack() was called in handler and HandlerFunc returns error,
+// msg.Nack() will be not sent because Ack was already sent.
+//
+// HandlerFunc's are executed parallel when multiple messages was received
+// (because msg.Ack() was sent in HandlerFunc or Subscriber supports multiple consumers).
type HandlerFunc func(msg *Message) ([]*Message, error)
+// HandlerMiddleware allows us to write something like decorators to HandlerFunc.
+// It can execute something before handler (for example: modify consumed message)
+// or after (modify produced messages, ack/nack on consumed message, handle errors, logging, etc.).
+//
+// It can be attached to the router by using `AddMiddleware` method.
+//
+// Example:
+// func ExampleMiddleware(h message.HandlerFunc) message.HandlerFunc {
+// return func(message *message.Message) ([]*message.Message, error) {
+// fmt.Println("executed before handler")
+// producedMessages, err := h(message)
+// fmt.Println("executed after handler")
+//
+// return producedMessages, err
+// }
+// }
type HandlerMiddleware func(h HandlerFunc) HandlerFunc
+// RouterPlugin is function which is executed on Router start.
type RouterPlugin func(*Router) error
type RouterConfig struct {
+ // CloseTimeout determines how long router should work for handlers when closing.
CloseTimeout time.Duration
}
@@ -94,6 +121,20 @@ func (r *Router) AddPlugin(p ...RouterPlugin) {
r.plugins = append(r.plugins, p...)
}
+// AddHandler adds a new handler.
+//
+// handlerName must be unique. For now, it is used only for debugging.
+//
+// subscribeTopic is a topic from which handler will receive messages.
+//
+// publishTopic is a topic to which router will produce messages retuened by handlerFunc.
+// When handler needs to publish to multiple topics,
+// it is recommended to just inject Publisher to Handler or implement middleware
+// which will catch messages and publish to topic based on metadata for example.
+//
+// pubSub is PubSub from which messages will be consumed and to which created messages will be published.
+// If you have separated Publisher and Subscriber object,
+// you can create PubSub object by calling message.NewPubSub(publisher, subscriber).
func (r *Router) AddHandler(
handlerName string,
subscribeTopic string,
@@ -110,6 +151,15 @@ func (r *Router) AddHandler(
return nil
}
+// AddNoPublisherHandler adds a new handler.
+// This handler cannot return messages.
+// When message is returned it will occur an error and Nack will be sent.
+//
+// handlerName must be unique. For now, it is used only for debugging.
+//
+// subscribeTopic is a topic from which handler will receive messages.
+//
+// subscriber is Subscriber from which messages will be consumed.
func (r *Router) AddNoPublisherHandler(
handlerName string,
subscribeTopic string,
@@ -138,6 +188,10 @@ func (r *Router) AddNoPublisherHandler(
return nil
}
+// Run runs all plugins and handlers and starts subscribing to provided topics.
+// This call is blocking until router is running.
+//
+// To stop Run() you should call Close() on the router.
func (r *Router) Run() (err error) {
if r.isRunning {
return errors.New("router is already running")
@@ -205,7 +259,10 @@ func (r *Router) Run() (err error) {
// Running is closed when router is running.
// In other words: you can wait till router is running using
-// <- r.Running()
+// fmt.Println("Starting router")
+// go r.Run()
+// <- r.Running()
+// fmt.Println("Router is running")
func (r *Router) Running() chan struct{} {
return r.running
}
diff --git a/message/subscriber.go b/message/subscriber.go
index b2324f5db..b3183f0b4 100644
--- a/message/subscriber.go
+++ b/message/subscriber.go
@@ -1,10 +1,17 @@
package message
type subscriber interface {
+ // Subscribe returns output channel with messages from provided topic.
+ // Channel is closed, when Close() was called to the subscriber.
+ //
+ // To receive next message, `Ack()` must be called on the received message.
+ // If message processing was failed and message should be redelivered `Nack()` should be called.
Subscribe(topic string) (chan *Message, error)
}
type Subscriber interface {
subscriber
+
+ // Close closes all subscriptions with their output channels and flush offsets etc. when needed.
Close() error
}
diff --git a/netlify.toml b/netlify.toml
new file mode 100644
index 000000000..61315ea79
--- /dev/null
+++ b/netlify.toml
@@ -0,0 +1,17 @@
+[build]
+ command = "hugo version && ./build.sh"
+ base = "docs/"
+ publish = "docs/public/"
+
+[context.production.environment]
+ HUGO_VERSION = "0.52"
+ HUGO_ENV = "production"
+ HUGO_ENABLEGITINFO = "true"
+
+[context.deploy-preview.environment]
+ HUGO_VERSION = "0.52"
+ HUGO_ENABLEGITINFO = "true"
+
+[context.branch-deploy.environment]
+ HUGO_VERSION = "0.52"
+ HUGO_ENABLEGITINFO = "true"
+
+### Configuration
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="type RouterConfig struct {" last_line_contains="RouterConfig) Validate()" padding_after="2" %}}
+{{% /render-md %}}
+
+### Handler
+
+At the beginning we need to implement HandlerFunc:
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// HandlerFunc is" last_line_contains="type HandlerFunc func" padding_after="1" %}}
+{{% /render-md %}}
+
+Next we need to add a new handler with `Router.AddHandler`:
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// AddHandler" last_line_contains=") error" padding_after="0" %}}
+{{% /render-md %}}
+
+And example usage from [Getting Started]({{< ref "/docs/getting-started#using-messages-router" >}}):
+{{% render-md %}}
+{{% load-snippet-partial file="content/docs/getting-started/router/main.go" first_line_contains="if err := router.AddHandler(" last_line_contains="panic(err)" padding_after="1" %}}
+{{% /render-md %}}
+
+### No publisher handler
+
+Not every handler needs to publish messages.
+You can add this kind of handler by using `Router.AddNoPublisherHandler`:
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// AddNoPublisherHandler" last_line_contains=") error" padding_after="0" %}}
+{{% /render-md %}}
+
+### Ack
+
+You don't need to call `msg.Ack()` or `msg.Nack()` after a message is processed (but you can, of course).
+`msg.Ack()` is called when `HanderFunc` doesn't return error. If the error was returned, `msg.Nack()` will be called.
+
+### Producing messages
+
+When returning multiple messages in the router,
+you should be aware that most of Publisher's implementations don't support [atomically publishing of the messages]({{< ref "/docs/pub-sub#publishing-multiple-messages" >}}).
+
+It may lead to producing only part of the messages and sending `msg.Nack()` when broker or storage is not available.
+
+When it is a problem, you should consider publishing maximum one message with one handler.
+
+### Running router
+
+To run the router, you need to call `Run()`.
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// Run" last_line_contains="func (r *Router) Run() (err error) {" padding_after="0" %}}
+{{% /render-md %}}
+
+#### Ensuring that router is running
+
+Sometimes, you want to do something after the router was started. You can use `Running()` method for this.
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// Running" last_line_contains="func (r *Router) Running()" padding_after="0" %}}
+{{% /render-md %}}
+
+### Execution model
+
+Some *Consumers* may support an only single stream of messages - that means that until `msg.Ack()` is sent you will not receive more messages.
+
+However, some *Consumers* can, for example, subscribe to multiple partitions in parallel and multiple messages will be sent even previous was not Acked (Kafka Consumer for example).
+The router can handle this case and spawn multiple HandlerFunc in parallel.
+
+
+### Middleware
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// HandlerMiddleware" last_line_contains="type HandlerMiddleware" padding_after="1" %}}
+{{% /render-md %}}
+
+A full list of standard middlewares can are in [message/router/middleware](https://github.com/ThreeDotsLabs/watermill/tree/master/message/router/middleware).
+
+### Plugin
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// RouterPlugin" last_line_contains="type RouterPlugin" padding_after="1" %}}
+{{% /render-md %}}
+
+A full list of standard plugins can are in [message/router/plugin](https://github.com/ThreeDotsLabs/watermill/tree/master/message/router/plugin).
diff --git a/docs/content/docs/pub-sub-implementations.md b/docs/content/docs/pub-sub-implementations.md
new file mode 100644
index 000000000..bd2524763
--- /dev/null
+++ b/docs/content/docs/pub-sub-implementations.md
@@ -0,0 +1,175 @@
++++
+title = "Pub/Sub's implementations"
+description = "Golang channel, Kafka, HTTP, Google Cloud Pub/Sub and more!"
+date = 2018-12-05T12:47:48+01:00
+weight = -800
+draft = false
+bref = "Golang channel, Kafka, HTTP, Google Cloud Pub/Sub and more!"
+toc = false
++++
+
+| Name | Publisher | Subscriber | Status |
+|------|-----------|------------|--------|
+| [Golang Channel]({{< ref "#golang-channel" >}}) | x | x | `prod-ready` |
+| [Kafka]({{< ref "#kafka" >}}) | x | x | `prod-ready` |
+| [HTTP]({{< ref "#http" >}}) | | x | `prod-ready` |
+| [Google Cloud Pub/Sub]({{< ref "#google-cloud-pub-sub" >}}) | x | x | [`in-development`](https://github.com/ThreeDotsLabs/watermill/pull/10) |
+| MySQL Binlog | | x | [`idea`](https://github.com/ThreeDotsLabs/watermill/issues/5) |
+
+All built-in implementations can be found in [message/infrastructure](https://github.com/ThreeDotsLabs/watermill/tree/master/message/infrastructure).
+
+### Golang Channel
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// GoChannel" last_line_contains="type GoChannel struct {" %}}
+{{% /render-md %}}
+
+#### Characteristics
+
+| Feature | Implements | Note |
+| ------- | ---------- | ---- |
+| ConsumerGroups | no | |
+| ExactlyOnceDelivery | yes | |
+| GuaranteedOrder | yes | |
+| Persistent | no| |
+
+##### Configuration
+
+You can inject configuration via the constructor.
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="func NewGoChannel" last_line_contains="logger:" %}}
+{{% /render-md %}}
+
+#### Publishing
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// Publish" last_line_contains="func (g *GoChannel) Publish" %}}
+{{% /render-md %}}
+
+#### Subscribing
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// Subscribe" last_line_contains="func (g *GoChannel) Subscribe" %}}
+{{% /render-md %}}
+
+#### Marshaler
+
+No marshaling is needed when sending messages within the process.
+
+### Kafka
+
+Kafka is one of the most popular Pub/Subs. We are providing Pub/Sub implementation based on [Confluent's bindings to `librdkafka`](https://github.com/confluentinc/confluent-kafka-go).
+
+`librdkafka` is required to run Kafka Pub/Sub. Installation guide can be found in [Getting Started]({{< ref "/docs/getting-started#subscribing_kafka" >}}).
+
+#### Characteristics
+
+| Feature | Implements | Note |
+| ------- | ---------- | ---- |
+| ConsumerGroups | yes | |
+| ExactlyOnceDelivery | no | in theory can be achieved with [Transactions](https://www.confluent.io/blog/transactions-apache-kafka/), currently no support for any Golang client |
+| GuaranteedOrder | yes | require [paritition key usage](#using-partition-key) |
+| Persistent | yes| |
+
+#### Configuration
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/subscriber.go" first_line_contains="type SubscriberConfig struct" last_line_contains="func (c SubscriberConfig)" %}}
+{{% /render-md %}}
+
+##### Passing custom `librdkafka` config
+
+You can pass custom config parameters (for example SSL Configuration) via `KafkaConfigOverwrite` in `SubscriberConfig` and `kafkaConfigOverwrite` to `NewPublisher`.
+
+You can find a list of available options in [librdkafka documentation](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md).
+
+#### Publishing
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/publisher.go" first_line_contains="// Publish" last_line_contains="func (p confluentPublisher) Publish" %}}
+{{% /render-md %}}
+
+#### Subscribing
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/subscriber.go" first_line_contains="// Subscribe" last_line_contains="func (s *confluentSubscriber) Subscribe" %}}
+{{% /render-md %}}
+
+#### Marshaler
+
+Watermill's messages cannot be directly sent to Kafka - they need to be marshaled. You can implement your marshaler or use default implementation.
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/marshaler.go" first_line_contains="// Marshaler" last_line_contains="func (DefaultMarshaler)" padding_after="0" %}}
+{{% /render-md %}}
+
+#### Partitioning
+
+Our Publisher has support for the partitioning mechanism.
+
+It can be done with special Marshaler implementation:
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/marshaler.go" first_line_contains="type kafkaJsonWithPartitioning" last_line_contains="func (j kafkaJsonWithPartitioning) Marshal" padding_after="0" %}}
+{{% /render-md %}}
+
+When using, you need to pass your function to generate partition key.
+It's a good idea to pass this partition key with metadata to not unmarshal entire message.
+
+{{< highlight >}}
+marshaler := kafka.NewWithPartitioningMarshaler(func(topic string, msg *message.Message) (string, error) {
+ return msg.Metadata.Get("partition"), nil
+})
+{{< /highlight >}}
+
+### HTTP
+
+For this moment only HTTP subscriber is available. There is an issue for a [HTTP publisher](https://github.com/ThreeDotsLabs/watermill/issues/17).
+
+HTTP subscriber allows us to send messages received by HTTP request (for example - webhooks).
+You can then post them to any Publisher. Here is an example with [sending HTTP messages to Kafka](https://github.com/ThreeDotsLabs/watermill/blob/master/_examples/http-to-kafka/main.go).
+
+When implemented, HTTP publisher can be used as webhooks sender.
+
+#### Characteristics
+
+| Feature | Implements | Note |
+| ------- | ---------- | ---- |
+| ConsumerGroups | no | |
+| ExactlyOnceDelivery | no | |
+| GuaranteedOrder | yes | |
+| Persistent | no| |
+
+#### Configuration
+
+The configuration of HTTP subscriber is done via the constructor.
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/http/subscriber.go" first_line_contains="// NewSubscriber" last_line_contains="func NewSubscriber(" %}}
+{{% /render-md %}}
+
+#### Running
+
+To run HTTP subscriber you need to run `StartHTTPServer()`. It needs to be run after `Subscribe()`.
+
+When using with the router, you should wait for the router to start.
+
+{{< highlight >}}
+<-r.Running()
+httpSubscriber.StartHTTPServer()
+{{< /highlight >}}
+
+#### Subscribing
+
+{{% render-md %}}
+{{% load-snippet-partial file="content/src-link/message/infrastructure/http/subscriber.go" first_line_contains="// Subscribe adds" last_line_contains="func (s *Subscriber) Subscribe" %}}
+{{% /render-md %}}
+
+### Google Cloud Pub/Sub
+
+In progress.
+
+### Implementing your own Pub/Sub
+
+There aren't your Pub/Sub implementation? Please check [Implementing custom Pub/Sub]({{< ref "pub-sub-implementing" >}}).
diff --git a/docs/content/docs/pub-sub-implementing.md b/docs/content/docs/pub-sub-implementing.md
new file mode 100644
index 000000000..29dc15595
--- /dev/null
+++ b/docs/content/docs/pub-sub-implementing.md
@@ -0,0 +1,39 @@
++++
+title = "Implementing custom Pub/Sub"
+description = "Bring Your Own Pub/Sub"
+date = 2018-12-05T12:48:34+01:00
+weight = -300
+draft = false
+bref = "Bring Your Own Pub/Sub"
+toc = true
++++
+
+### Pub/Sub interface
+
+In simple words - to implement Pub/Sub you need to implement `message.PubSub` interface.
+
+{{% render-md %}}
+{{% load-snippet file="content/src-link/message/publisher.go" %}}
+{{% load-snippet file="content/src-link/message/subscriber.go" %}}
+{{% load-snippet-partial file="content/src-link/message/pubsub.go" first_line_contains="type PubSub interface {" last_line_contains="Close() error" padding_after="1" %}}
+{{% /render-md %}}
+
+### TODO list
+
+But they are some things about which you cannot forget:
+
+1. Good logging
+2. Replaceable and configureable messages marshaler
+3. `Close()` implementation for publisher and subscriber which is:
+ - idempotent
+ - working event when publisher or subscriber is blocked (for example: when waiting for Ack)
+ - working when subscriber output channel is blocked (because nothing is listening for it)
+4. `Ack()` **and** `Nack()` support for consumed messages
+5. Redelivery on `Nack()` on consumed message
+6. Use [Universal Pub/Sub tests]({{< ref "/docs/pub-sub#universal-tests" >}})
+7. Performance optimizations
+8. godoc's, [Markdown docs]({{< ref "/docs/pub-sub-implementations" >}}) and [examples Getting Started](/docs/getting-started)
+
+We will be also thankful for submitting [merge requests](https://github.com/ThreeDotsLabs/watermill/pulls) with new Pub/Subs implementation.
+
+If anything is not clear feel free to use any of our [support channels]({{< ref "/support" >}}), we will we'll be glad to help.
diff --git a/docs/content/docs/pub-sub.md b/docs/content/docs/pub-sub.md
new file mode 100644
index 000000000..f8492e416
--- /dev/null
+++ b/docs/content/docs/pub-sub.md
@@ -0,0 +1,80 @@
++++
+title = "Pub/Sub"
+description = "Publishers and Subscribers"
+date = 2018-12-05T12:47:30+01:00
+weight = -900
+draft = false
+bref = "Publishers and Subscribers"
+toc = true
++++
+
+### Publisher
+
+{{% render-md %}}
+{{% load-snippet file="content/src-link/message/publisher.go" %}}
+{{% /render-md %}}
+
+#### Publishing multiple messages
+
+Most publishers implementations don't support atomic publishing of messages.
+That means, that when publishing one of the messages failed the next messages will be not published.
+
+#### Async publish
+
+Publish can be synchronous or asynchronous - it depends on implementation.
+
+#### `Close()`
+
+Close should flush unsent messages if the publisher is async.
+**It's important to not forget to close subscriber**, in the other hand you may lose some of the messages.
+
+### Subscriber
+
+{{% render-md %}}
+{{% load-snippet file="content/src-link/message/subscriber.go" %}}
+{{% /render-md %}}
+
+#### Ack/Nack mechanism
+
+It is *Subscriber's* responsibility to handle `Ack` and `Nack` from a message.
+A proper implementation should wait for `Ack` or `Nack` before consuming the next message.
+
+**Important Subscriber's implementation notice**:
+Ack/offset to message's storage/broker **must** be sent after Ack from Watermill's message.
+If it wouldn't, it is a possibility to lose messages when the process will die before messages were processed.
+
+#### `Close()`
+
+Close closes all subscriptions with their output channels and flush offsets etc. when needed.
+
+### At-least-once delivery
+
+Watermill is build with [at-least-once delivery](http://www.cloudcomputingpatterns.org/at_least_once_delivery/) semantics.
+That means, that when some error with occur when processing message and Ack cannot be sent the message will be redelivered.
+
+You need to keep it in mind and build your application to be [idempotent](http://www.cloudcomputingpatterns.org/idempotent_processor/) or implement deduplication mechanism.
+
+Unfortunately, it's not possible to create universal [*middleware*]({{< ref "/docs/messages-router#middleware" >}}) for deduplication but we encourage you to make your own.
+
+### Universal tests
+
+Every Pub/Sub is similar.
+To don't implement separated tests for every Pub/Sub we create test suite which should be passed by any Pub/Sub implementation.
+
+These tests can be found in `message/infrastructure/test_pubsub.go`.
+
+### Built-in implementations
+
+To check available Pub/Subs implementation please check [Pub/Sub's implementations]({{< ref "/docs/pub-sub-implementations" >}})
+
+### Implementing custom Pub/Sub
+
+When any implementation of Pub/Sub. [Implementing custom Pub/Sub]({{< ref "/docs/pub-sub-implementing" >}}).
+
+We will be also thankful for submitting [merge requests](https://github.com/ThreeDotsLabs/watermill/pulls) with new Pub/Subs implementation.
+
+You can also request new Pub/Sub implementation by submitting a [new issue](https://github.com/ThreeDotsLabs/watermill/issues).
+
+### Keep going!
+
+When you already know, how Pub/Sub is working we recommend to check [*Message Router component*]({{< ref "/docs/messages-router" >}}).
diff --git a/docs/content/docs/troubleshooting.md b/docs/content/docs/troubleshooting.md
new file mode 100644
index 000000000..f97aaa09e
--- /dev/null
+++ b/docs/content/docs/troubleshooting.md
@@ -0,0 +1,66 @@
++++
+title = "Troubleshooting"
+description = "When something went wrong"
+weight = 0
+draft = false
+toc = true
+bref = "When something went wrong"
+type = "docs"
++++
+
+### Logging
+
+In most cases, you will find the answer to your problems in a log.
+Watermill offers a significant amount of logs of different severity levels.
+
+If you are using `StdLoggerAdapter`, just change `debug`, and `trace` options to true:
+
+{{< highlight >}}
+logger := watermill.NewStdLogger(true, true)
+{{< /highlight >}}
+
+### I had a deadlock
+
+When running locally, you can run:
+
+- `CTRL + \` for Linux
+- `kill -s SIGQUIT [pid]` for another UNIX systems
+
+
+to send `SIGQUIT` to the process.
+It will kill the process and print all goroutines with the line on which they are now.
+
+```
+SIGQUIT: quit
+PC=0x45e7c3 m=0 sigcode=128
+
+goroutine 1 [runnable]:
+github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).sendMessage(0xc000024100, 0x7c5250, 0xd, 0xc000872d70, 0x0, 0x0)
+ /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:83 +0x36a
+github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).Publish(0xc000024100, 0x7c5250, 0xd, 0xc000098530, 0x1, 0x1, 0x0, 0x0)
+ /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:53 +0x6d
+main.publishMessages(0x7fdf7a317000, 0xc000024100)
+ /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:43 +0x1ec
+main.main()
+ /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:36 +0x20a
+
+// ...
+```
+
+When running in production, and you don't want to kill the entire process it is a good idea to use [pprof](https://golang.org/pkg/net/http/pprof/).
+
+You can visit [http://localhost:6060/debug/pprof/goroutine?debug=1](http://localhost:6060/debug/pprof/goroutine?debug=1) to find all goroutines status.
+
+
+```
+goroutine profile: total 5
+1 @ 0x41024c 0x6a8311 0x6a9bcb 0x6a948d 0x7028bc 0x70260a 0x42f187 0x45c971
+# 0x6a8310 github.com/ThreeDotsLabs/watermill.LogFields.Add+0xd0 /home/example/go/src/github.com/ThreeDotsLabs/watermill/log.go:15
+# 0x6a9bca github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).sendMessage+0x6fa /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:75
+# 0x6a948c github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).Publish+0x6c /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:53
+# 0x7028bb main.publishMessages+0x1eb /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:43
+# 0x702609 main.main+0x209 /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:36
+# 0x42f186 runtime.main+0x206 /usr/lib/go/src/runtime/proc.go:201
+
+// ...
+```
\ No newline at end of file
diff --git a/docs/content/support.md b/docs/content/support.md
new file mode 100644
index 000000000..5039a5f81
--- /dev/null
+++ b/docs/content/support.md
@@ -0,0 +1,14 @@
++++
+title = "Support"
+description = ""
++++
+
+### Community Support
+
+Please join us on the `#watermill` channel on the [Gophers slack](https://gophers.slack.com/): You can get the invite [here](https://gophersinvite.herokuapp.com/).
+
+### Professional Support
+
+For enterprise support please contact us by e-mail: **{{% contact-email %}}**.
+
+You can also use the [contact form](https://threedotslabs.com/#contact-form) on our website.
diff --git a/docs/layouts/index.html b/docs/layouts/index.html
new file mode 100644
index 000000000..448fdbfd9
--- /dev/null
+++ b/docs/layouts/index.html
@@ -0,0 +1,65 @@
+{{ define "title"}} {{ .Site.Title}} {{end}}
+{{ define "header"}} {{ partial "header" .}} {{end}}
+{{ define "main"}}
+
+
+
+