MessageContentTypeEnforcer { public byte[] enforceAvro(String payloadContentType, byte[] data, Schema schema, Topic topic) { String contentTypeLowerCase = StringUtils.lowerCase(payloadContentType); if (isJSON(contentTypeLowerCase)) { return defaultJsonAvroconverter.convertToAvro(data, schema); } else if (isAvroJSON(contentTypeLowerCase)) { return avroEncodedJsonAvroConverter.convertToAvro(data, schema); } else if (isAvroBinary(contentTypeLowerCase)) { return data; } else { throw new UnsupportedContentTypeException(payloadContentType, topic); } }  byte[] enforceAvro(String payloadContentType, byte[] data, Schema schema, Topic topic);  }
@Test public void shouldConvertToAvroWhenReceivedJSONOnAvroTopic() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("application/json", avroMessage.asJson().getBytes(), schema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test public void sh1ouldConvertToAvroWhenReceivedJSONOnAvroTopic() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("application/json", avroMessage.asJson().getBytes(), testSchema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test public void shouldConvertToAvroWhenReceivedAvroJSONOnAvroTopic() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("avro/json", avroMessage.asAvroEncodedJson().getBytes(), schema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test public void shouldStringContentTypeOfAdditionalOptionsWhenInterpretingIt() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("application/json;encoding=utf-8", avroMessage.asJson().getBytes(), schema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test public void shouldNotConvertWhenReceivingAvroOnAvroTopic() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("avro/binary", avroMessage.asBytes(), schema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test public void shouldBeCaseInsensitiveForPayloadContentType() throws IOException { byte[] enforcedMessage = enforcer.enforceAvro("AVRO/Binary", avroMessage.asBytes(), schema.getSchema(), topic); assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); }
@Test(expected = UnsupportedContentTypeException.class) public void shouldThrowUnsupportedContentTypeExceptionWhenReceivedWrongContentType() throws IOException { enforcer.enforceAvro(MediaType.TEXT_PLAIN, avroMessage.asBytes(), schema.getSchema(), topic); }