Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add support for zstd compression #695

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Add support for zstd compression #695

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3a550ce
Add support for zstd compression
abitofevrything Sep 6, 2024
5c4f7cd
Add comment
abitofevrything Sep 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions lib/src/api_options.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,8 @@ class GatewayApiOptions extends RestApiOptions {
Map<String, String> get gatewayConnectionOptions => {
'v': apiVersion.toString(),
'encoding': payloadFormat.value,
if (compression == GatewayCompression.transport) 'compress': 'zlib-stream',
if (compression == GatewayCompression.transportZLib) 'compress': 'zlib-stream',
if (compression == GatewayCompression.transportZStd) 'compress': 'zstd-stream',
};

/// Create a new [GatewayApiOptions].
Expand All @@ -108,7 +109,7 @@ class GatewayApiOptions extends RestApiOptions {
super.userAgent,
required this.intents,
this.payloadFormat = GatewayPayloadFormat.json,
this.compression = GatewayCompression.transport,
this.compression = GatewayCompression.transportZLib,
this.shards,
this.totalShards,
this.largeThreshold,
Expand Down Expand Up @@ -138,11 +139,17 @@ enum GatewayCompression {
/// No compression is used.
none,

/// The entire connection is compressed.
transport,
/// The entire connection is compressed with ZLib.
transportZLib,

/// The entire connection is compressed with ZStd
transportZStd,

/// Each packet is individually compressed.
///
/// Cannot be used if [GatewayPayloadFormat.etf] is used.
payload,
payload;

@Deprecated('Use GatewayCompression.transportZLib for ZLib transport compression.')
static const transport = GatewayCompression.transportZLib;
}
9 changes: 7 additions & 2 deletions lib/src/gateway/shard_runner.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import 'dart:convert';
import 'dart:io';
import 'dart:math';

import 'package:es_compression/zstd.dart';
import 'package:eterl/eterl.dart';
import 'package:nyxx/src/api_options.dart';
import 'package:nyxx/src/errors.dart';
Expand Down Expand Up @@ -367,7 +368,8 @@ class ShardConnection extends Stream<GatewayEvent> implements StreamSink<Send> {
final connection = await WebSocket.connect(gatewayUri);

final uncompressedStream = switch (runner.data.apiOptions.compression) {
GatewayCompression.transport => decompressTransport(connection.cast<List<int>>()),
GatewayCompression.transportZLib => decompressZLibTransport(connection.cast<List<int>>()),
GatewayCompression.transportZStd => decompressZStdTransport(connection.cast<List<int>>()),
GatewayCompression.payload => decompressPayloads(connection),
GatewayCompression.none => connection,
};
Expand Down Expand Up @@ -450,7 +452,7 @@ class ShardConnection extends Stream<GatewayEvent> implements StreamSink<Send> {
Future<void> get done => websocket.done.then((_) => _sentController.done);
}

Stream<dynamic> decompressTransport(Stream<List<int>> raw) {
Stream<dynamic> decompressZLibTransport(Stream<List<int>> raw) {
final filter = RawZLibFilter.inflateFilter();

return raw.map((chunk) {
Expand All @@ -465,6 +467,9 @@ Stream<dynamic> decompressTransport(Stream<List<int>> raw) {
});
}

// See https://github.com/instantiations/es_compression/issues/52 for why the isNotEmpty check is needed.
Stream<dynamic> decompressZStdTransport(Stream<List<int>> raw) => raw.transform(zstd.decoder).where((message) => message.isNotEmpty);
Comment on lines +470 to +471
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is isNotEmpty still needed? instantiations/es_compression#52 (comment)


Stream<dynamic> decompressPayloads(Stream<dynamic> raw) => raw.map((message) {
if (message is String) {
return message;
Expand Down
2 changes: 1 addition & 1 deletion pubspec.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,11 @@ dependencies:
http: ^1.0.0
logging: ^1.0.1
path: ^1.8.3
retry: ^3.1.0
eterl: ^1.0.1
runtime_type: ^1.0.1
meta: ^1.9.1
oauth2: ^2.0.2
es_compression: ^2.0.12

dev_dependencies:
test: ^1.22.0
Expand Down
64 changes: 15 additions & 49 deletions test/integration/gateway_integration_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,55 +19,21 @@ void main() {
await expectLater(client.close(), completes);
}

test(
'JSON (uncompressed)',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: GatewayCompression.none,
payloadFormat: GatewayPayloadFormat.json,
)),
);

test(
'JSON (payload compression)',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: GatewayCompression.payload,
payloadFormat: GatewayPayloadFormat.json,
)),
);

test(
'JSON (transport compression)',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: GatewayCompression.transport,
payloadFormat: GatewayPayloadFormat.json,
)),
);

test(
'ETF (uncompressed)',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: GatewayCompression.none,
payloadFormat: GatewayPayloadFormat.etf,
)),
);

test(
'ETF (transport compression)',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: GatewayCompression.transport,
payloadFormat: GatewayPayloadFormat.etf,
)),
);
for (final compression in GatewayCompression.values) {
for (final payloadFormat in GatewayPayloadFormat.values) {
if (payloadFormat != GatewayPayloadFormat.json && compression == GatewayCompression.payload) continue;

test(
'${payloadFormat.value.toUpperCase()} (${compression.name})',
() => testClient(GatewayApiOptions(
token: testToken!,
intents: GatewayIntents.none,
compression: compression,
payloadFormat: payloadFormat,
)),
);
}
}

test('Multiple shards', () async {
const shardCount = 5;
Expand Down
Loading