SQL Publisher/Subscriber #56
Conversation
…gs, but consumer groups should work now
| ) | ||
| require.NoError(t, err) | ||
|
|
||
| ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) |
There was a problem hiding this comment.
the cancel function returned by context.WithTimeout should be called, not discarded, to avoid a context leak (from govet)
Rework Publisher to use this adapter. Subscriber pending.
| ) | ||
| require.NoError(t, err) | ||
|
|
||
| ctx, _ := context.WithTimeout(context.Background(), 15*time.Second) |
There was a problem hiding this comment.
the cancel function returned by context.WithTimeout should be called, not discarded, to avoid a context leak (from govet)
dev/events.sql
Outdated
| @@ -0,0 +1,17 @@ | |||
| CREATE TABLE IF NOT EXISTS messages ( | |||
message/infrastructure/sql/schema.go
Outdated
| Logger watermill.LoggerAdapter | ||
| } | ||
|
|
||
| func (s *DefaultSchema) EnsureTableForTopicQueries(topic string) []string { |
There was a problem hiding this comment.
I'm not sure about this name, maybe something more related to schema, intiialiation etc
There was a problem hiding this comment.
done 74d84d8
SchemaInitializingQueries
message/infrastructure/sql/schema.go
Outdated
| } | ||
|
|
||
| func (s *DefaultSchema) InsertQuery(topic string) string { | ||
| if s.Logger == nil { |
There was a problem hiding this comment.
private getter will be nice to handle it universally
There was a problem hiding this comment.
no longer applies, schema doesn't log anymore, only publisher/subscriber
message/infrastructure/sql/schema.go
Outdated
| "`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,", | ||
| "`payload` json DEFAULT NULL,", | ||
| "`metadata` json DEFAULT NULL,", | ||
| "`topic` varchar(255) NOT NULL", |
message/infrastructure/sql/schema.go
Outdated
|
|
||
| func (s *DefaultSchema) InsertArgs(topic string, msg *message.Message) (args []interface{}, err error) { | ||
| logger := s.Logger | ||
| defer func() { |
There was a problem hiding this comment.
IMO we should not return error and log it, and logging should be also done in "generic part"
There was a problem hiding this comment.
moved to publisher/subscriber 74d84d8
message/infrastructure/sql/schema.go
Outdated
| `(uuid, payload, metadata, topic) VALUES (?,?,?,?)`, | ||
| }, " ") | ||
|
|
||
| s.Logger.Info("Preparing query to insert messages", watermill.LogFields{ |
There was a problem hiding this comment.
couldn't all logging be done in general implementation?
There was a problem hiding this comment.
moved to publisher/subscriber 74d84d8
message/infrastructure/sql/schema.go
Outdated
| }() | ||
|
|
||
| var uuid ulid.ULID | ||
| uuid, err = ulid.Parse(msg.UUID) |
There was a problem hiding this comment.
what do you think to make it configurable? because it can be different to everyone (and not much people are using ulid)
There was a problem hiding this comment.
oooor we may store uuid as string?
There was a problem hiding this comment.
b29759f no longer requiring ulid, just string
| }, nil | ||
| } | ||
|
|
||
| // Publish inserts the messages as rows into the MessagesTable. |
There was a problem hiding this comment.
maybe some info that transaction should be handled by the user of Publish (and examples in docs?)
| if c.ResendInterval == 0 { | ||
| c.ResendInterval = time.Second | ||
| } | ||
| if c.MessagesTable == "" { |
There was a problem hiding this comment.
isn't it duplicated with SchemaAdapter?
There was a problem hiding this comment.
removed from both subscriber and publisher
message/infrastructure/sql/topic.go
Outdated
|
|
||
| // sanitizeTopicName checks if the topic name contains any characters which could be unsuitable for the SQL Pub/Sub. | ||
| // Topics are translated into SQL tables and patched into some queries, so this is done to prevent injection as well. | ||
| func sanitizeTopicName(topic string) error { |
There was a problem hiding this comment.
isn't it validation, sanitization for me is more about changing
|
|
||
| err := s.query(ctx, topic, out, logger) | ||
| if err != nil { | ||
| logger.Error("Error querying for message", err, nil) |
| // go on querying | ||
| } | ||
|
|
||
| err := s.query(ctx, topic, out, logger) |
There was a problem hiding this comment.
how it's working with reconnecting
There was a problem hiding this comment.
I'm thinking about implementing the reconnect test later on
| // it is finalized after the ACK is written | ||
| var tx *sql.Tx | ||
| tx, err = s.db.BeginTx(ctx, &sql.TxOptions{ | ||
| //Isolation: sql.LevelSerializable, |
Subscriber works by querying the messages table with SELECT in configurable intervals