@Test public void serializerWithEncodingStructuredAndFormat() { CloudEventSerializer serializer = new CloudEventSerializer(); EventFormatProvider.getInstance().registerFormat(CSVFormat.INSTANCE); HashMap<String, Object> config = new HashMap<>(); config.put(CloudEventSerializer.ENCODING_CONFIG, "STRUCTURED"); config.put(CloudEventSerializer.EVENT_FORMAT_CONFIG, CSVFormat.INSTANCE.serializedContentType()); assertThatCode(() -> serializer.configure(config, false)) .doesNotThrowAnyException(); assertProducesMessagesWithEncoding(serializer, Encoding.STRUCTURED); }
@Override public void configure(Map<String, ?> configs, boolean isKey) { if (isKey) { throw new IllegalArgumentException("Cannot use CloudEventSerializer as key serializer"); } Object encodingConfig = configs.get(ENCODING_CONFIG); if (encodingConfig instanceof String) { this.encoding = Encoding.valueOf((String) encodingConfig); } else if (encodingConfig instanceof Encoding) { this.encoding = (Encoding) encodingConfig; } else if (encodingConfig != null) { throw new IllegalArgumentException(ENCODING_CONFIG + " can be of type String or " + Encoding.class.getCanonicalName()); } if (this.encoding == Encoding.UNKNOWN) { throw new IllegalArgumentException(ENCODING_CONFIG + " cannot be " + Encoding.UNKNOWN); } if (this.encoding == Encoding.STRUCTURED) { Object eventFormatConfig = configs.get(EVENT_FORMAT_CONFIG); if (eventFormatConfig instanceof String) { this.format = EventFormatProvider.getInstance().resolveFormat((String) eventFormatConfig); if (this.format == null) { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be resolved with " + eventFormatConfig); } } else if (eventFormatConfig instanceof EventFormat) { this.format = (EventFormat) eventFormatConfig; } else { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be null and can be of type String or " + EventFormat.class.getCanonicalName()); } } }
CloudEventSerializer implements Serializer<CloudEvent> { @Override public void configure(Map<String, ?> configs, boolean isKey) { if (isKey) { throw new IllegalArgumentException("Cannot use CloudEventSerializer as key serializer"); } Object encodingConfig = configs.get(ENCODING_CONFIG); if (encodingConfig instanceof String) { this.encoding = Encoding.valueOf((String) encodingConfig); } else if (encodingConfig instanceof Encoding) { this.encoding = (Encoding) encodingConfig; } else if (encodingConfig != null) { throw new IllegalArgumentException(ENCODING_CONFIG + " can be of type String or " + Encoding.class.getCanonicalName()); } if (this.encoding == Encoding.UNKNOWN) { throw new IllegalArgumentException(ENCODING_CONFIG + " cannot be " + Encoding.UNKNOWN); } if (this.encoding == Encoding.STRUCTURED) { Object eventFormatConfig = configs.get(EVENT_FORMAT_CONFIG); if (eventFormatConfig instanceof String) { this.format = EventFormatProvider.getInstance().resolveFormat((String) eventFormatConfig); if (this.format == null) { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be resolved with " + eventFormatConfig); } } else if (eventFormatConfig instanceof EventFormat) { this.format = (EventFormat) eventFormatConfig; } else { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be null and can be of type String or " + EventFormat.class.getCanonicalName()); } } } }
CloudEventSerializer implements Serializer<CloudEvent> { @Override public void configure(Map<String, ?> configs, boolean isKey) { if (isKey) { throw new IllegalArgumentException("Cannot use CloudEventSerializer as key serializer"); } Object encodingConfig = configs.get(ENCODING_CONFIG); if (encodingConfig instanceof String) { this.encoding = Encoding.valueOf((String) encodingConfig); } else if (encodingConfig instanceof Encoding) { this.encoding = (Encoding) encodingConfig; } else if (encodingConfig != null) { throw new IllegalArgumentException(ENCODING_CONFIG + " can be of type String or " + Encoding.class.getCanonicalName()); } if (this.encoding == Encoding.UNKNOWN) { throw new IllegalArgumentException(ENCODING_CONFIG + " cannot be " + Encoding.UNKNOWN); } if (this.encoding == Encoding.STRUCTURED) { Object eventFormatConfig = configs.get(EVENT_FORMAT_CONFIG); if (eventFormatConfig instanceof String) { this.format = EventFormatProvider.getInstance().resolveFormat((String) eventFormatConfig); if (this.format == null) { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be resolved with " + eventFormatConfig); } } else if (eventFormatConfig instanceof EventFormat) { this.format = (EventFormat) eventFormatConfig; } else { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be null and can be of type String or " + EventFormat.class.getCanonicalName()); } } }  }
CloudEventSerializer implements Serializer<CloudEvent> { @Override public void configure(Map<String, ?> configs, boolean isKey) { if (isKey) { throw new IllegalArgumentException("Cannot use CloudEventSerializer as key serializer"); } Object encodingConfig = configs.get(ENCODING_CONFIG); if (encodingConfig instanceof String) { this.encoding = Encoding.valueOf((String) encodingConfig); } else if (encodingConfig instanceof Encoding) { this.encoding = (Encoding) encodingConfig; } else if (encodingConfig != null) { throw new IllegalArgumentException(ENCODING_CONFIG + " can be of type String or " + Encoding.class.getCanonicalName()); } if (this.encoding == Encoding.UNKNOWN) { throw new IllegalArgumentException(ENCODING_CONFIG + " cannot be " + Encoding.UNKNOWN); } if (this.encoding == Encoding.STRUCTURED) { Object eventFormatConfig = configs.get(EVENT_FORMAT_CONFIG); if (eventFormatConfig instanceof String) { this.format = EventFormatProvider.getInstance().resolveFormat((String) eventFormatConfig); if (this.format == null) { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be resolved with " + eventFormatConfig); } } else if (eventFormatConfig instanceof EventFormat) { this.format = (EventFormat) eventFormatConfig; } else { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be null and can be of type String or " + EventFormat.class.getCanonicalName()); } } }  @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, CloudEvent data); @Override byte[] serialize(String topic, Headers headers, CloudEvent data); }
CloudEventSerializer implements Serializer<CloudEvent> { @Override public void configure(Map<String, ?> configs, boolean isKey) { if (isKey) { throw new IllegalArgumentException("Cannot use CloudEventSerializer as key serializer"); } Object encodingConfig = configs.get(ENCODING_CONFIG); if (encodingConfig instanceof String) { this.encoding = Encoding.valueOf((String) encodingConfig); } else if (encodingConfig instanceof Encoding) { this.encoding = (Encoding) encodingConfig; } else if (encodingConfig != null) { throw new IllegalArgumentException(ENCODING_CONFIG + " can be of type String or " + Encoding.class.getCanonicalName()); } if (this.encoding == Encoding.UNKNOWN) { throw new IllegalArgumentException(ENCODING_CONFIG + " cannot be " + Encoding.UNKNOWN); } if (this.encoding == Encoding.STRUCTURED) { Object eventFormatConfig = configs.get(EVENT_FORMAT_CONFIG); if (eventFormatConfig instanceof String) { this.format = EventFormatProvider.getInstance().resolveFormat((String) eventFormatConfig); if (this.format == null) { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be resolved with " + eventFormatConfig); } } else if (eventFormatConfig instanceof EventFormat) { this.format = (EventFormat) eventFormatConfig; } else { throw new IllegalArgumentException(EVENT_FORMAT_CONFIG + " cannot be null and can be of type String or " + EventFormat.class.getCanonicalName()); } } }  @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, CloudEvent data); @Override byte[] serialize(String topic, Headers headers, CloudEvent data); final static String ENCODING_CONFIG; final static String EVENT_FORMAT_CONFIG; }