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
e183bd8
commit 3752d5e
Showing
3 changed files
with
183 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| 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 QueueConnectionListener struct { | ||
| name string | ||
| cfg queue.Cfg | ||
| queue queue.Queue | ||
|
|
||
| uuid uuid.UUID | ||
| registered bool | ||
| err error | ||
| mutex sync.Mutex | ||
| done chan struct{} | ||
| } | ||
|
|
||
| func NewQueueConnectionListener(name string, cfg queue.Cfg) *QueueConnectionListener { | ||
| return &QueueConnectionListener{ | ||
| name: name, | ||
| cfg: cfg, | ||
| done: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (l *QueueConnectionListener) Added(conn *tarantool.Connection, | ||
| role connection_pool.Role) { | ||
| if role != connection_pool.MasterRole { | ||
| return | ||
| } | ||
|
|
||
| l.mutex.Lock() | ||
| defer l.mutex.Unlock() | ||
|
|
||
| if l.err != nil { | ||
| return | ||
| } | ||
|
|
||
| l.queue = queue.New(conn, l.name) | ||
| if !l.registered { | ||
| opts := queue.CfgOpts{InReplicaset: true, Ttr: 60 * time.Second} | ||
| if l.err = l.queue.Cfg(opts); l.err == nil { | ||
| if l.uuid, l.err = l.queue.Identify(nil); l.err == nil { | ||
| l.err = l.queue.Create(l.cfg) | ||
| } | ||
| } | ||
| l.registered = l.err == nil | ||
| close(l.done) | ||
| } else { | ||
| l.queue.Identify(&l.uuid) | ||
| l.queue.Create(l.cfg) | ||
| } | ||
| } | ||
|
|
||
| func (l *QueueConnectionListener) Removed(conn *tarantool.Connection, | ||
| role connection_pool.Role) { | ||
| return | ||
| } | ||
|
|
||
| func (l *QueueConnectionListener) Updated(conn *tarantool.Connection, | ||
| oldRole, newRole connection_pool.Role) { | ||
| return | ||
| } | ||
|
|
||
| // Example demonstrates how to use the queue package with the connection_pool | ||
| // package. First of all, you need to create a connection listener for the | ||
| // a ConnectionPool object to process new connections from RW-instances. | ||
| // | ||
| // You need to registry 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 QueueConnectionListener.Added() 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, | ||
| }, | ||
| } | ||
| l := NewQueueConnectionListener("test_queue", cfg) | ||
| poolOpts := connection_pool.OptsPool{ | ||
| CheckTimeout: 1 * time.Second, | ||
| ConnectionListener: l, | ||
| } | ||
| connPool, err := connection_pool.ConnectWithOpts(servers, connOpts, poolOpts) | ||
| if err != nil { | ||
| log.Fatalf("unable to connect to the pool: %s", err) | ||
| } | ||
|
|
||
| defer connPool.Close() | ||
|
|
||
| <-l.done | ||
|
|
||
| if l.err != nil { | ||
| log.Fatalf("unable to identify in the pool: %s", l.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