Rewrite gateway emitter for significant performance improvement

[zeylahellyer]

Rewrite the Emitter for performance. The emitter is what's called when bytes, an event, or a JSON payload is being emitted by the ShardProcessor.

Emitters have internally held a DashMap of listeners; shards and clusters could create multiple streams of events via Cluster::events, Cluster::some_events, Shard::events, and Shard::some_events. This was an intentional design decision, but over time we've seen that this capability hasn't seen much purpose.

Instead, we can create a stream of events when clusters or shards are created, and simply return that from Cluster::new, Shard::new, and their builder equivalents. This means that instead of only returning the cluster or shard, a two element tuple is returned of the cluster or shard and the stream of events. Events can then be sent by the ShardProcessor directly over a channel that the stream polls. This results in removing DashMap as a dependency and a significant performance improvement, as seen in the benchmark below.

The first pair of tests is emitting the raw bytes of a GUILD_ROLE_DELETE dispatch payload. The first test has a listener that doesn't want to receive the EventTypeFlags::SHARD_PAYLOAD variant, while the second does. While the first test sees no change - the change tends to go up and down due to how small of a scale of time it is - the second sees a significant improvement.

The second pair of tests is similar to the first, but instead of emitting the raw bytes of the payload it is attempting to emit a received payload as a deserialized event. If a listener does not want an event type, then the payload is discarded. If the user does want the event type, then the payload is deserialized and then sent to the user. Much like the first pair, the first test receives little improvement, but the second receives a comparable performance improvement.

emit bytes: doesn't want
                        time:   [3.1906 ns 3.1919 ns 3.1936 ns]
                        change: [+0.1098% +0.3127% +0.5112%] (p = 0.00 < 0.05)
                        Change within noise threshold.
Found 1 outliers among 100 measurements (1.00%)
  1 (1.00%) high mild

emit bytes: wants       time:   [77.733 ns 78.493 ns 79.427 ns]
                        change: [-79.574% -79.312% -79.025%] (p = 0.00 < 0.05)
                        Performance has improved.
Found 11 outliers among 100 measurements (11.00%)
  2 (2.00%) low severe
  3 (3.00%) low mild
  4 (4.00%) high mild
  2 (2.00%) high severe

emit json: doesn't want time:   [7.2541 ns 7.2587 ns 7.2656 ns]
                        change: [-2.2163% -2.1091% -1.9993%] (p = 0.00 < 0.05)
                        Performance has improved.
Found 11 outliers among 100 measurements (11.00%)
  3 (3.00%) high mild
  8 (8.00%) high severe

emit json: wants        time:   [274.46 ns 276.97 ns 279.86 ns]
                        change: [-60.786% -59.956% -58.741%] (p = 0.00 < 0.05)
                        Performance has improved.
Found 5 outliers among 100 measurements (5.00%)
  2 (2.00%) high mild
  3 (3.00%) high severe

Criterion benchmark used:

use criterion::{criterion_group, criterion_main, Criterion};

use twilight_gateway::{shard::emitter::Emitter, EventTypeFlags};

const INPUT: &str = r##"{
    "op": 0,
    "s": 2,
    "d": {
        "guild_id": "1",
        "role_id": "2"
    },
    "t": "GUILD_ROLE_DELETE"
}"##;
const INPUT_BYTES: &[u8] = INPUT.as_bytes();

fn emit_bytes(emitter: &Emitter) {
    emitter.bytes(INPUT_BYTES);
}

fn emit_json(emitter: &Emitter, input: &mut str) {
    emitter.json(0, Some(12), Some("GUILD_ROLE_DELETE"), input).unwrap();
}

fn criterion_benchmark(c: &mut Criterion) {
    c.bench_function("emit bytes: doesn't want", |b| {
        let (emitter, _rx) = Emitter::new(EventTypeFlags::empty());

        b.iter(|| emit_bytes(&emitter))
    });

    c.bench_function("emit bytes: wants", |b| {
        let (emitter, _rx) = Emitter::new(EventTypeFlags::SHARD_PAYLOAD);

        b.iter(|| emit_bytes(&emitter))
    });

    c.bench_function("emit json: doesn't want", |b| {
        let (emitter, _rx) = Emitter::new(EventTypeFlags::empty());
        let mut input = INPUT.to_owned();

        b.iter(|| emit_json(&emitter, &mut input));
    });

    c.bench_function("emit json: wants", |b| {
        let (emitter, _rx) = Emitter::new(EventTypeFlags::ROLE_DELETE);
        let mut input = INPUT.to_owned();

        b.iter(|| emit_json(&emitter, &mut input));
    });
}

criterion_group!(benches, criterion_benchmark);
criterion_main!(benches);

[Erk-]

Looks like a good and solid pr just a really minor nit that I spotted

[zeylahellyer]

Addressed comments