|
1 | 1 | const test = require('brittle') |
2 | 2 | const b4a = require('b4a') |
| 3 | +const compact = require('compact-encoding') |
| 4 | +const DebuggingStream = require('debugging-stream') |
3 | 5 | const NoiseSecretStream = require('@hyperswarm/secret-stream') |
| 6 | +const Protomux = require('protomux') |
4 | 7 | const { |
5 | 8 | create, |
6 | 9 | createStored, |
@@ -757,6 +760,61 @@ test('multiplexing with createProtocolStream (ondiscoverykey is called)', async |
757 | 760 | s2.destroy() |
758 | 761 | }) |
759 | 762 |
|
| 763 | +test('custom bulk channel does not delay hypercore discovery', async function (t) { |
| 764 | + const advertised = await create(t) |
| 765 | + |
| 766 | + const n1 = new NoiseSecretStream(true) |
| 767 | + const n2 = new NoiseSecretStream(false) |
| 768 | + |
| 769 | + const slowIncoming = new DebuggingStream(n2.rawStream, { |
| 770 | + writeSpeed: 512 * 1024, |
| 771 | + readSpeed: Infinity |
| 772 | + }) |
| 773 | + |
| 774 | + t.teardown(function () { |
| 775 | + n1.destroy() |
| 776 | + n2.destroy() |
| 777 | + slowIncoming.destroy() |
| 778 | + }) |
| 779 | + |
| 780 | + n1.rawStream.pipe(slowIncoming).pipe(n1.rawStream) |
| 781 | + |
| 782 | + const mux1 = Protomux.from(n1) |
| 783 | + const mux2 = Protomux.from(n2) |
| 784 | + |
| 785 | + const app1 = mux1.createChannel({ protocol: 'app/bulk' }) |
| 786 | + const app2 = mux2.createChannel({ protocol: 'app/bulk' }) |
| 787 | + const bulk = app1.addMessage({ encoding: compact.buffer }) |
| 788 | + app2.addMessage({ encoding: compact.buffer }) |
| 789 | + |
| 790 | + app1.open() |
| 791 | + app2.open() |
| 792 | + |
| 793 | + await Promise.all([app1.fullyOpened(), app2.fullyOpened()]) |
| 794 | + |
| 795 | + // Model a user-defined high-throughput app channel sharing the same |
| 796 | + // Hyperswarm connection as Hypercore. It writes 64 * 64 KiB = 4 MiB at |
| 797 | + // 512 KiB/s before Hypercore opens its replication channel, so the |
| 798 | + // receiver can take ~8s to discover the core behind those app bytes. |
| 799 | + bulk.send(Buffer.alloc(64 * 64 * 1024)) |
| 800 | + |
| 801 | + const discovery = new Promise((resolve) => { |
| 802 | + Hypercore.createProtocolStream(n2, { |
| 803 | + ondiscoverykey(discoveryKey) { |
| 804 | + if (b4a.equals(discoveryKey, advertised.discoveryKey)) resolve('discovered') |
| 805 | + } |
| 806 | + }) |
| 807 | + }) |
| 808 | + |
| 809 | + const stream = Hypercore.createProtocolStream(n1) |
| 810 | + const replication = advertised.replicate(stream, { keepAlive: false }) |
| 811 | + |
| 812 | + t.teardown(() => replication.destroy()) |
| 813 | + |
| 814 | + const result = await Promise.race([discovery, sleep(5000).then(() => 'timeout')]) |
| 815 | + t.is(result, 'discovered', 'hypercore discovery should not wait behind app bulk data') |
| 816 | +}) |
| 817 | + |
760 | 818 | test('seeking while replicating', async function (t) { |
761 | 819 | const a = await create(t) |
762 | 820 | const b = await create(t, a.key) |
@@ -2958,4 +3016,8 @@ async function waitForRequestBlock(core) { |
2958 | 3016 | } |
2959 | 3017 | } |
2960 | 3018 |
|
| 3019 | +function sleep(ms) { |
| 3020 | + return new Promise((resolve) => setTimeout(resolve, ms)) |
| 3021 | +} |
| 3022 | + |
2961 | 3023 | function noop() {} |
0 commit comments