forked from fl00r/go-tarantool-1.6
-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
queue: add an example with connection_pool
The example demonstrates how to use the queue package with the connection_pool package. Closes #176
- Loading branch information
1 parent
3a0809c
commit d0314fe
Showing
4 changed files
with
191 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| package queue_test | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/google/uuid" | ||
| "github.com/tarantool/go-tarantool" | ||
| "github.com/tarantool/go-tarantool/connection_pool" | ||
| "github.com/tarantool/go-tarantool/queue" | ||
| ) | ||
|
|
||
| type QueueConnectionHandler struct { | ||
| name string | ||
| cfg queue.Cfg | ||
|
|
||
| uuid uuid.UUID | ||
| registered bool | ||
| err error | ||
| registerMutex sync.Mutex | ||
| done chan struct{} | ||
| } | ||
|
|
||
| var _ connection_pool.ConnectionHandler = &QueueConnectionHandler{} | ||
|
|
||
| func NewQueueConnectionHandler(name string, cfg queue.Cfg) *QueueConnectionHandler { | ||
| return &QueueConnectionHandler{ | ||
| name: name, | ||
| cfg: cfg, | ||
| done: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (h *QueueConnectionHandler) Discovered(conn *tarantool.Connection, | ||
| role connection_pool.Role) error { | ||
| // The queue only works with a master instance. | ||
| if role != connection_pool.MasterRole { | ||
| return nil | ||
| } | ||
|
|
||
| // We need to register a shared session only once. | ||
| h.registerMutex.Lock() | ||
|
|
||
| if h.err != nil { | ||
| h.registerMutex.Unlock() | ||
| return h.err | ||
| } | ||
|
|
||
| q := queue.New(conn, h.name) | ||
| if !h.registered { | ||
| opts := queue.CfgOpts{InReplicaset: true, Ttr: 60 * time.Second} | ||
| if h.err = q.Cfg(opts); h.err == nil { | ||
| if h.uuid, h.err = q.Identify(nil); h.err == nil { | ||
| h.err = q.Create(h.cfg) | ||
| } | ||
| } | ||
| h.registered = h.err == nil | ||
| close(h.done) | ||
| h.registerMutex.Unlock() | ||
| return h.err | ||
| } else { | ||
| // We already registered. We don't need the lock any more. | ||
| h.registerMutex.Unlock() | ||
| if _, err := q.Identify(&h.uuid); err != nil { | ||
| return err | ||
| } | ||
| if err := q.Create(h.cfg); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (h *QueueConnectionHandler) Deactivated(conn *tarantool.Connection, | ||
| role connection_pool.Role) error { | ||
| return nil | ||
| } | ||
|
|
||
| // Example demonstrates how to use the queue package with the connection_pool | ||
| // package. First of all, you need to create a connection handler for the | ||
| // a ConnectionPool object to process new connections from RW-instances. | ||
| // | ||
| // You need to register a shared session UUID at a first master connection. | ||
| // It needs to be used to re-identify as the shared session on new | ||
| // RW-instances. See QueueConnectionHandler.Discovered() implementation. | ||
| // | ||
| // After that, you need to create a ConnectorAdapter object with RW mode for | ||
| // the ConnectionPool to send requests into RW-instances. This adapter can | ||
| // be used to create a ready-to-work queue object. | ||
| func Example_connectionPool() { | ||
| servers := []string{ | ||
| "127.0.0.1:3014", | ||
| "127.0.0.1:3015", | ||
| "127.0.0.1:3016", | ||
| } | ||
| connOpts := tarantool.Opts{ | ||
| Timeout: 500 * time.Millisecond, | ||
| User: "test", | ||
| Pass: "test", | ||
| } | ||
|
|
||
| cfg := queue.Cfg{ | ||
| Temporary: false, | ||
| IfNotExists: true, | ||
| Kind: queue.FIFO, | ||
| Opts: queue.Opts{ | ||
| Ttl: 10 * time.Second, | ||
| }, | ||
| } | ||
| h := NewQueueConnectionHandler("test_queue", cfg) | ||
| poolOpts := connection_pool.OptsPool{ | ||
| CheckTimeout: 1 * time.Second, | ||
| ConnectionHandler: h, | ||
| } | ||
| connPool, err := connection_pool.ConnectWithOpts(servers, connOpts, poolOpts) | ||
| if err != nil { | ||
| log.Fatalf("unable to connect to the pool: %s", err) | ||
| } | ||
|
|
||
| defer connPool.Close() | ||
|
|
||
| <-h.done | ||
|
|
||
| if h.err != nil { | ||
| log.Fatalf("unable to identify in the pool: %s", h.err) | ||
| } | ||
|
|
||
| rw := connection_pool.NewConnectorAdapter(connPool, connection_pool.RW) | ||
| q := queue.New(rw, "test_queue") | ||
|
|
||
| fmt.Println("A Queue object is ready to work.") | ||
|
|
||
| if _, err = q.Put("test_data"); err != nil { | ||
| log.Fatalf("unable to put data into the queue: %s", err) | ||
| } | ||
|
|
||
| if task, err := q.Take(); err != nil { | ||
| log.Fatalf("unable to take data from the queue: %s", err) | ||
| } else { | ||
| task.Ack() | ||
| fmt.Println("data:", task.Data()) | ||
| } | ||
| // Output: | ||
| // A Queue object is ready to work. | ||
| // data: test_data | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters