Skip to content
This repository was archived by the owner on Jun 23, 2023. It is now read-only.

Commit fd06df4

Browse files
authored
Merge pull request #19 from heetch/bootstap-documentation
Bootstrap documentation
2 parents 4c461b5 + d1e7c9a commit fd06df4

File tree

11 files changed

+212
-17
lines changed

11 files changed

+212
-17
lines changed

README.md

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,3 +10,10 @@ Felice is very much a work in progress. We expect to move forward quickly, but
1010
## Why "Felice"?
1111
Felice Bauer was, at one time, Franz Kafka's fiance. He wrote her many messages, which she faithfuly kept, and later published.
1212

13+
## Where should I start?
14+
If you wish to send messages via Kafka, you should start by reading
15+
the documentation for the `producer` package. If you wish to consume
16+
messages from Kafka, you should start by reading the documentation for
17+
the `consumer` package. The `message` package contains the `Message` type that is
18+
shared by both `consumer` and `producer` code.
19+

consumer/consumer.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,10 @@ type clusterConsumer interface {
2222
Close() error
2323
}
2424

25-
// Consumer is a Kafka consumer.
25+
// Consumer is the structure used to consume messages from Kafka.
26+
// Having constructed a Conusmer you should use its Handle method to
27+
// register per-topic handlers and, finally, call it's Serve method to
28+
// begin consuming messages.
2629
type Consumer struct {
2730
RetryInterval time.Duration
2831
Metrics MetricsReporter
@@ -81,7 +84,12 @@ func newClusterConfig(clientID string) *cluster.Config {
8184
return c
8285
}
8386

84-
// Serve runs the consumer and listens for new messages on the given topics.
87+
// Serve runs the consumer and listens for new messages on the given
88+
// topics. You must provide it with unique clientID and the address
89+
// of one or more Kafka brokers. Serve will block until it is
90+
// instructed to stop, which you can achieve by calling Consumer.Stop.
91+
// When Serve terminates it will return an Error or nil to indicate
92+
// that it excited without error.
8593
func (c *Consumer) Serve(clientID string, addrs ...string) error {
8694
c.setup()
8795

consumer/consumer_internal_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -344,7 +344,7 @@ type testHandler struct {
344344

345345
// HandleMessage will keep a count of how many times it is called and,
346346
// if a testCase is set on the testHandler, it will run it with the
347-
// message that HandleMessage recieved, allowing us to make assertions
347+
// message that HandleMessage received, allowing us to make assertions
348348
// about the nature of that message.
349349
func (h *testHandler) HandleMessage(m *message.Message) error {
350350
h.CallCount++

consumer/doc.go

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
// Consumer is Felice's primary entrance point for receiving messages
2+
// from a Kafka cluster.
3+
//
4+
// There is no special construction function for the Consumer
5+
// structure as all of its public members are optional, and we shall
6+
// discuss them below. Thus you construct a Consumer by the normal Go
7+
// means:
8+
//
9+
// c := felice.Consumer{}
10+
//
11+
// Once you've constructed a consumer you must add message handlers to
12+
// it. This is done by calling the Consumer.Handle method. Each time
13+
// you call Handle you'll pass a topic name and a type that implements
14+
// the handler.Handler interface. Their can only ever be one handler
15+
// associated with a topic so, if you call Handle multiple times with
16+
// the same topic, they will update the handler registered for the
17+
// topic, and only the final one will count. A typical call to Handle
18+
// looks like this:
19+
//
20+
// c.Handle("testmsg", handler.HandlerFunc(func(m *message.Message) error {
21+
// // Do something of your choice here!
22+
// return nil // .. or return an actual error.
23+
// }))
24+
//
25+
// Once you've registered all your handlers you may call
26+
// Consumer.Serve. Serve requires a client ID and a slice of strings,
27+
// each of which is the address of a Kafka broker to attempt to
28+
// communicate with. Serve will start a go routine for each partition
29+
// to consume messages and pass them to their per-topic
30+
// handlers. Serve itself will block until Consumer.Stop is called.
31+
// When Serve terminates it will return an error, which will be nil
32+
// under normal circumstances.
33+
//
34+
// Note that any calls to Consumer.Handle after
35+
// Consumer.Serve has been called will have no effect.
36+
//
37+
// Tweaking the consumer
38+
// --------------------------
39+
// The first public member is the RetryInterval, a time.Duration that
40+
// controls how long the Felice consumer will wait before trying to
41+
// consume a message from Kafka that failed the first time around.
42+
// The default value if 1 second.
43+
//
44+
// The second public member is Metrics. Metrics stores a type that
45+
// implements the felice.MetricsReporter interface. If you provide an
46+
// implementation, then it's Report function will be called every time
47+
// a message is successfully handled. The Report function will
48+
// receive a copy of the message.Message that was handled, along with
49+
// a map[string]string containing metrics about the handling of the
50+
// message. Currently we pass the following metrics: "attempts" (the
51+
// number of attempts it took before the message was handled);
52+
// "msgOffset" (the marked offset for the message on the topic); and
53+
// "remainingOffset" (the difference between the high water mark and
54+
// the current offset).
55+
package consumer

consumer/handler/collection.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,8 @@ func (h *Collection) Topics() []string {
4545
}
4646

4747
// Set associates the given Handler to the given Topic within the
48-
// collection. If a Handler was already associated with the Topic,
49-
// then that association will be lost and replaced by the new one. It
48+
// collection. If a Handler was already associated with the Topic,
49+
// then that association will be lost and replaced by the new one. It
5050
// is safe to use Set from concurrent code.
5151
func (h *Collection) Set(topic string, handler Handler) {
5252
h.Lock()

consumer/handler/doc.go

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
// The handler package defines the mechanisms by which a received
2+
// message can be handled, and how handlers themselves can be managed.
3+
// The most obvious way to use the handler package is via the consumer
4+
// package, but this isn't an absolute requirement. We do however
5+
// require that the messages handled are felice's message.Message
6+
// type.
7+
//
8+
// The Handler interface defines the signature for all felice
9+
// Handlers. There are two common ways to comply with this interface.
10+
// The first is simply to create a type with the HandleMessage
11+
// function:
12+
//
13+
// type MyFooHandler struct { }
14+
//
15+
// func (mfh MyFooHandler) HandleMessage(msg *message.Message) error {
16+
// fmt.Printf("%+v", *msg)
17+
// }
18+
//
19+
// This approach has the advantage of not actually requiring you to
20+
// import the handler package when defining handlers for use with the
21+
// consumer.Consumer.
22+
//
23+
// The second approach is to cast a function to the HandlerFunc type
24+
// defined in this package:
25+
//
26+
// h := handler.HandlerFunc(func(msg *message.Message) error {
27+
// fmt.Printf("%+v", *msg)
28+
// })
29+
//
30+
// Handlers can also be managed in Collections. The Collection struct
31+
// will allow exactly one handler to be associated with each topic,
32+
// and will, on demand return a list of all topics for which a handler
33+
// has been registered. For example:
34+
//
35+
// col := handler.Collection{}
36+
//
37+
// col.Set("my-topic", handler.HandlerFunc(func(msg *message.Message) error) {
38+
// fmt.Printf("Got message on my-topic: %+v", *msg)
39+
// return nil
40+
// })
41+
//
42+
// col.Set("your-topic", handler.HandlerFunc(func(msg *message.Message) error) {
43+
// fmt.Printf("Got message on your-topic: %+v", msg)
44+
// return nil
45+
// })
46+
//
47+
// yourHandler, ok := col.Get("your-topic")
48+
// if !ok {
49+
// fmt.Println("Couldn't find a handler for your-topic")
50+
// }
51+
//
52+
// for _, t := range col.Topics {
53+
// fmt.Printf("We have a handler for: %s", t)
54+
// }
55+
//
56+
package handler

consumer/handler/func.go

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,13 @@ import (
44
"github.com/heetch/felice/message"
55
)
66

7-
// A HandlerFunc is a function type that mimics the interface of the
8-
// HandleMessage function in the Handler interface. If you cast a
9-
// function to this type, it will comply with the Handler interface.
7+
// A HandlerFunc is a function that supports the Handler interface.
8+
// Calls to Handler.HandleMessage made against any function cast to
9+
// this type will result in the function itself being called.
1010
type HandlerFunc func(*message.Message) error
1111

12+
// HandleMessage implements the Handler interface by calling the
13+
// HandlerFunc to which it is associated.
1214
func (h HandlerFunc) HandleMessage(msg *message.Message) error {
1315
return h(msg)
1416
}

consumer/handler/interface.go

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,22 @@ package handler
22

33
import "github.com/heetch/felice/message"
44

5-
// A Handler handles a Message.
5+
// Handler is the interface for handling consumed messages. You can
6+
// either add a compliant HandleMessage function to some type by hand,
7+
// or make use of the HandlerFunc. Note that felice Handlers receive
8+
// felice Message types.
69
type Handler interface {
10+
// HandleMessage functions will receive a *message.Message
11+
// type, and may do with it what they wish. However, some
12+
// caution should be used when returning an error from a
13+
// HandleMessage function. Errors are considered an
14+
// indication that the message has not been handled. After
15+
// waiting a respectable amount of time, the HandleMessage
16+
// function will be called again with the same message. This
17+
// will continue until on of the following conditions is met:
18+
//
19+
// - 1. HandleMessage returns nil, instead of an error.
20+
// - 2. A system administrator intervenes.
21+
// - 3. The Earth is consumed by the sun.
722
HandleMessage(*message.Message) error
823
}

message/doc.go

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
// The message package contains the Message type. When using
2+
// the felice Producer, Message will be the type you send.
3+
// When using the felice Consumer, you will register handlers that
4+
// receive the Message type.
5+
//
6+
// You can create a new Message by calling New:
7+
//
8+
// msg := New("my-topic", "simple string message")
9+
//
10+
// The value passed as the 2nd argument can be any Go type that can be
11+
// marshalled with encoding/json. Two default headers are added to
12+
// all Messages:
13+
//
14+
// - Message-Id : a universally unique ID for the message
15+
// - Produced-At : the current time in the UTC timezone.
16+
//
17+
// New can also be passed zero, one or many additional Options. An
18+
// Option is a function that receives a pointer to the Message and can
19+
// modify it directly prior to it being returned by New. Two
20+
// predefined Options exists in felice: Header and Key.
21+
//
22+
// For example, if you want to create a Message with a custom header
23+
// field you could request it as follows:
24+
//
25+
// msg := New("my-topic", "cold potatoes ain't hot!", Header("subject", "potatoes"))
26+
//
27+
package message

message/message.go

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,17 +9,23 @@ import (
99
uuid "github.com/satori/go.uuid"
1010
)
1111

12-
// Message contains informations about the message to be sent to Kafka.
12+
// Message represents a message to be sent via Kafka, or received from
13+
// it. When using Felice's Consumer, any Handlers that you register
14+
// will receive Messages as they're arguments. When using the Felice
15+
// Producer, you will be sending Messages. When making a Message to
16+
// be sent it is essential that you use the New function to do so.
1317
type Message struct {
14-
// Kafka topic.
18+
// The Kafka topic this Message applies to.
1519
Topic string
1620

1721
// If specified, messages with the same key will be sent to the same Kafka partition.
1822
Key string
1923

20-
// Body of the Kafka message.
24+
// Body of the Kafka message. For now this will always be a
25+
// JSON marshaled form of whatever was passed to New.
2126
Body []byte
2227

28+
// The time at which this Message was produced.
2329
ProducedAt time.Time
2430

2531
// Partition where this publication was stored.
@@ -31,11 +37,20 @@ type Message struct {
3137
// Headers of the message.
3238
Headers map[string]string
3339

34-
// Unique id of the message.
40+
// Unique ID of the message.
3541
ID string
3642
}
3743

38-
// New creates a new configured message.
44+
// New creates a new Message, correctly configured for use with the
45+
// felice Producer. You should not attempt to create Message types by
46+
// hand (unless you really know what you're doing! - I'm just some
47+
// documentation, not the police).
48+
//
49+
// Two headers will be added to all Messages:
50+
//
51+
// - Message-Id : a universally unique ID for the message
52+
// - Produced-At : the current time in the UTC timezone.
53+
//
3954
func New(topic string, value interface{}, opts ...Option) (*Message, error) {
4055
if topic == "" {
4156
return nil, fmt.Errorf("messages require a non-empty topic")

0 commit comments

Comments
 (0)