smocket/src/ack-native.test.ts at main · electrohyun/smocket · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import { expect, it, vi } from 'vitest';
import { Server } from './mock-server';
import { count, observeDisconnect, receive, track } from './test-events';

it('discards an ack retained after an outgoing observer disconnects the client', async () => {
  const io = new Server('http://ack-outgoing-disconnect');
  const client = io.connect();
  const serverSocket = await io.nextConnection();
  const senderAnswers = count(client, 'sender-answer');
  const retained = new Promise<(value: string) => void>((resolve) => {
    client.on('held', (ack: (value: string) => void) => resolve(ack));
  });
  serverSocket.onAnyOutgoing((event) => {
    if (event === 'held') client.disconnect();
  });
  const { disconnected } = observeDisconnect(serverSocket);

  serverSocket.emit('held', () => {
    serverSocket.emit('sender-answer');
  });
  const answer = await retained;
  await disconnected;
  answer('late');

  const connected = receive(client, 'connect');
  const nextServerSocket = io.nextConnection();
  client.connect();
  const reconnected = await nextServerSocket;
  reconnected.on('ack-marker', (ack: () => void) => ack());
  await connected;
  await client.emitWithAck('ack-marker');

  expect(senderAnswers.count).toBe(0);
  await io.close();
});

it('skips a buffered timed promise settled during reconnect flush', async () => {
  vi.useFakeTimers();
  try {
    const io = new Server('http://ack-reconnect-flush');
    const client = io.connect();
    const serverSocket = await io.nextConnection();
    const { disconnected } = observeDisconnect(serverSocket);
    client.disconnect();
    await disconnected;

    client.onAnyOutgoing((event) => {
      if (event === 'advance-time-during-flush') vi.advanceTimersByTime(20);
    });
    client.emit('advance-time-during-flush');
    const pending = client.timeout(20).emitWithAck('settled-during-flush');

    const connected = receive(client, 'connect');
    const nextServerSocket = io.nextConnection();
    client.connect();
    const socket = await nextServerSocket;
    const delivered = track(socket, 'settled-during-flush');
    socket.on('ack-marker', (ack: () => void) => ack());
    await connected;
    await expect(pending).rejects.toThrow('operation has timed out');
    await client.emitWithAck('ack-marker');

    expect(delivered.received).toBe(false);
    await io.close();
  } finally {
    vi.useRealTimers();
  }
});

it('stops reconnect flushing when an outgoing observer disconnects', async () => {
  const io = new Server('http://ack-reconnect-flush-stop');
  const client = io.connect();
  const serverSocket = await io.nextConnection();
  const { disconnected } = observeDisconnect(serverSocket);
  client.disconnect();
  await disconnected;

  let interrupted = false;
  let laterObserved = false;
  client.onAnyOutgoing((event) => {
    if (event === 'interrupt-flush' && !interrupted) {
      interrupted = true;
      client.disconnect();
    }
    if (event === 'after-interruption') laterObserved = true;
  });
  client.emit('interrupt-flush');
  const response = new Promise<unknown[]>((resolve) => {
    client.timeout(60_000).emit('after-interruption', (...args: unknown[]) => {
      client.emit('callback-settled');
      resolve(args);
    });
  });

  const interruptedConnection = receive(client, 'disconnect');
  const firstReconnection = io.nextConnection();
  client.connect();
  await firstReconnection;
  await interruptedConnection;
  expect(laterObserved).toBe(false);

  const connected = receive(client, 'connect');
  const secondReconnection = io.nextConnection();
  client.connect();
  const reconnected = await secondReconnection;
  const deliveryOrder: string[] = [];
  const callbackSettlements = count(reconnected, 'callback-settled');
  reconnected.on('interrupt-flush', () => deliveryOrder.push('interrupt-flush'));
  reconnected.on('after-interruption', (ack: (value: string) => void) => {
    deliveryOrder.push('after-interruption');
    ack('answer');
  });
  reconnected.on('ack-marker', (ack: () => void) => ack());
  await connected;
  await expect(response).resolves.toEqual([null, 'answer']);
  await client.emitWithAck('ack-marker');

  expect(callbackSettlements.count).toBe(1);
  expect(deliveryOrder).toEqual(['interrupt-flush', 'after-interruption']);
  expect(laterObserved).toBe(true);
  await io.close();
});