NTable

A table data structure for NATS Streaming.

The primary use case for maintaining a lookup table of data derived from a stream. It does this by maintaining an internal key-value store. The default implementation uses an in-memory store, but stores are easy to implement since the interface is a basic key-value interface.

type Store interface {
  Get(key []byte) ([]byte, error)
  Set(key, val []byte) error
  Del(key []byte) error
}

Status

Design phase. Suggestions welcome.

Example

package main
import (
    "encoding/json"
    "fmt"
    "time"
    "github.com/bruth/ntable"
    "github.com/nats-io/go-nats-streaming"
)
// Record with an ID and some data.
type Record struct {
    ID   string
    Data string
}
func main() {
    conn, _ := stan.Connect("test-cluster", "test-client")
    t := &ntable.Table{
        Conn:    conn,
        Channel: "users",
        // Takes a message, derives the key and value and updates the store.
        Handle: func(s ntable.Store, m *stan.Msg) {
            var r Record
            json.Unmarshal(m.Data, &r)
            s.Set([]byte(r.ID), []byte(r.Data))
        },
    }
    // Open connection (subscription) to channel.
    t.Open()
    defer t.Close()
    // Publish a record about "pam" (likely happening from another process or thread).
    b, _ := json.Marshal(&Record{
        ID:   "pam",
        Data: "color=blue city=Philadelphia food=sushi",
    })
    conn.Publish("users", b)
    // A few moments pass..
    time.Sleep(5 * time.Millisecond)
    // We can get the data about pam using the table.
    val, _ := t.Get([]byte("pam"))
    fmt.Println(string(val))
}

License

MIT.