Register the broker wait before the first poll (closes #157)
check / check (push) Waiting to run
check / check (push) Waiting to run
GET /api/v1/messages read the client output queue and only then registered with the broker. Notify wakes only the channels registered at that moment, so a message queued between the empty read and the registration waited out the whole long-poll timeout. The handler now registers first and removes its channel on every exit path; the IRC relay loop registers before each drain in the same way. The handlers read the queue through a function field that defaults to Database.PollMessages, so tests can act just after a read: one checks that a message queued there comes back at once, others that no wait channel stays registered however a request ends. Model: opus-5-5
This commit was merged in pull request #174.
This commit is contained in:
@@ -27,17 +27,22 @@ func (c *Conn) relayMessages(ctx context.Context) {
|
||||
default:
|
||||
}
|
||||
|
||||
// Register with the broker before draining, so that a
|
||||
// message queued after the drain has read the queue
|
||||
// still signals waitCh.
|
||||
waitCh := c.brk.Wait(c.sessionID)
|
||||
|
||||
// Drain any available messages.
|
||||
delivered := c.drainQueue(ctx)
|
||||
|
||||
if delivered {
|
||||
c.brk.Remove(c.sessionID, waitCh)
|
||||
|
||||
// Tight loop while there are messages.
|
||||
continue
|
||||
}
|
||||
|
||||
// Wait for notification or timeout.
|
||||
waitCh := c.brk.Wait(c.sessionID)
|
||||
|
||||
select {
|
||||
case <-waitCh:
|
||||
// New message notification — loop back.
|
||||
|
||||
@@ -47,14 +47,13 @@ func (env *testEnv) queueMessage(
|
||||
env.brk.Notify(sessionID)
|
||||
}
|
||||
|
||||
// TestRelayKeepsOneWaitChannel checks that an idle IRC
|
||||
// listener connection holds at most one broker wait channel
|
||||
// across several relay ticks, and none once it closes.
|
||||
// TestRelayKeepsOneWaitChannel checks that an IRC listener
|
||||
// connection holds at most one broker wait channel across
|
||||
// several relay ticks, both while idle and after the relay
|
||||
// has delivered it a message, and none once it closes.
|
||||
func TestRelayKeepsOneWaitChannel(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const idleTicks = 5
|
||||
|
||||
env := newTestEnv(t)
|
||||
|
||||
client := env.dial(t)
|
||||
@@ -65,15 +64,17 @@ func TestRelayKeepsOneWaitChannel(t *testing.T) {
|
||||
t.Fatalf("get session of idle: %v", err)
|
||||
}
|
||||
|
||||
for tick := range idleTicks {
|
||||
time.Sleep(ircserver.PollInterval)
|
||||
env.checkOneWaitChannel(t, sessionID)
|
||||
|
||||
got := env.brk.ListenerCount(sessionID)
|
||||
if got > 1 {
|
||||
t.Fatalf("tick %d: got %d wait channels, want at most 1",
|
||||
tick+1, got)
|
||||
}
|
||||
}
|
||||
env.queueMessage(
|
||||
t, "idle", "PRIVMSG", "carol", "idle", []string{"hello"},
|
||||
)
|
||||
|
||||
client.readUntil(func(l string) bool {
|
||||
return strings.Contains(l, "hello")
|
||||
})
|
||||
|
||||
env.checkOneWaitChannel(t, sessionID)
|
||||
|
||||
err = client.conn.Close()
|
||||
if err != nil {
|
||||
@@ -92,6 +93,28 @@ func TestRelayKeepsOneWaitChannel(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// checkOneWaitChannel fails the test if the session
|
||||
// sessionID holds more than one broker wait channel at any of
|
||||
// several relay ticks.
|
||||
func (env *testEnv) checkOneWaitChannel(
|
||||
t *testing.T,
|
||||
sessionID int64,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
const ticks = 5
|
||||
|
||||
for tick := range ticks {
|
||||
time.Sleep(ircserver.PollInterval)
|
||||
|
||||
got := env.brk.ListenerCount(sessionID)
|
||||
if got > 1 {
|
||||
t.Fatalf("tick %d: got %d wait channels, want at most 1",
|
||||
tick+1, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestRelayMultiLineBody checks that each line of a PRIVMSG
|
||||
// or NOTICE body sent over the HTTP API reaches an IRC
|
||||
// listener user as its own line.
|
||||
|
||||
Reference in New Issue
Block a user