docs: public ByteBuf encode(GatewayMessage message) throws MessageCodecException {
      ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
      try (JsonGenerator generator =
          jsonFactory.createGenerator(
              (OutputStream) new ByteBufOutputStream(byteBuf), JsonEncoding.UTF8)) {
        generator.writeStartObject();
  
        // headers
        for (Entry<String, String> header : message.headers().entrySet()) {
          String fieldName = header.getKey();
          String value = header.getValue();
          switch (fieldName) {
            case STREAM_ID_FIELD:
            case SIGNAL_FIELD:
            case INACTIVITY_FIELD:
            case RATE_LIMIT_FIELD:
              generator.writeNumberField(fieldName, Long.parseLong(value));
              break;
            default:
              generator.writeStringField(fieldName, value);
          }
        }
  
        // data
        Object data = message.data();
        if (data != null) {
          if (data instanceof ByteBuf) {
            ByteBuf dataBin = (ByteBuf) data;
            if (dataBin.isReadable()) {
              try {
                generator.writeFieldName(DATA_FIELD);
                generator.writeRaw(":");
                generator.flush();
                byteBuf.writeBytes(dataBin);
              } finally {
                if (releaseDataOnEncode) {
                  ReferenceCountUtil.safestRelease(dataBin);
                }
              }
            }
          } else {
            generator.writeObjectField(DATA_FIELD, data);
          }
        }
  
        generator.writeEndObject();
      } catch (Throwable ex) {
        ReferenceCountUtil.safestRelease(byteBuf);
        Optional.ofNullable(message.data()).ifPresent(ReferenceCountUtil::safestRelease);
        LOGGER.error("Failed to encode message: {}", message, ex);
        throw new MessageCodecException("Failed to encode message", ex);
      }
      return byteBuf;
    }