@SuppressWarnings("unchecked")
    // visible for testing
    public StitchRequestBody createStitchRequestBody(final Message inMessage) {
        if (inMessage.getBody() instanceof StitchRequestBody) {
            return createStitchRequestBodyFromStitchRequestBody(inMessage.getBody(StitchRequestBody.class), inMessage);
        }

        if (inMessage.getBody() instanceof StitchMessage) {
            return createStitchRequestBodyFromStitchMessages(Collections.singletonList(inMessage.getBody(StitchMessage.class)),
                    inMessage);
        }

        if (inMessage.getBody() instanceof Iterable) {
            return createStitchRequestBodyFromIterable(inMessage.getBody(Iterable.class), inMessage);
        }

        if (inMessage.getBody() instanceof Map) {
            return createStitchRecordFromMap(inMessage.getBody(Map.class), inMessage);
        }

        throw new IllegalArgumentException("Message body data `" + inMessage.getBody() + "` type is not supported");
    }
@Test
    void testIfCreateFromIterable() {
        final StitchConfiguration configuration = new StitchConfiguration();
        configuration.setTableName("table_1");
        configuration.setStitchSchema(StitchSchema.builder().addKeyword("field_1", "string").build());
        configuration.setKeyNames("field_1");

        final StitchMessage stitchMessage1 = StitchMessage.builder()
                .withData("field_1", "stitchMessage1")
                .withSequence(1)
                .build();

        final StitchMessage stitchMessage2 = StitchMessage.builder()
                .withData("field_1", "stitchMessage2-1")
                .withData("field_2", "stitchMessage2-2")
                .withSequence(2)
                .build();

        final StitchRequestBody stitchMessage2RequestBody = StitchRequestBody.builder()
                .addMessage(stitchMessage2)
                .withSchema(StitchSchema.builder().addKeyword("field_1", "integer").build())
                .withTableName("table_1")
                .withKeyNames(Collections.singleton("field_1"))
                .build();

        final Map<String, Object> stitchMessage3 = new LinkedHashMap<>();
        stitchMessage3.put(StitchMessage.DATA, Collections.singletonMap("field_1", "stitchMessage3"));
        stitchMessage3.put(StitchMessage.SEQUENCE, 3L);

        final StitchMessage stitchMessage4 = StitchMessage.builder()
                .withData("field_1", "stitchMessage4")
                .withSequence(4)
                .build();

        final Exchange stitchMessage4Exchange = new DefaultExchange(context);
        stitchMessage4Exchange.getMessage().setBody(stitchMessage4);

        final StitchMessage stitchMessage5 = StitchMessage.builder()
                .withData("field_1", "stitchMessage5")
                .withSequence(5)
                .build();

        final Message stitchMessage5Message = new DefaultExchange(context).getMessage();
        stitchMessage5Message.setBody(stitchMessage5);

        final List<Object> inputMessages = new LinkedList<>();
        inputMessages.add(stitchMessage1);
        inputMessages.add(stitchMessage2RequestBody);
        inputMessages.add(stitchMessage3);
        inputMessages.add(stitchMessage4Exchange);
        inputMessages.add(stitchMessage5Message);

        final StitchProducerOperations operations = new StitchProducerOperations(new TestClient(), configuration);

        final Exchange exchange = new DefaultExchange(context);
        exchange.getMessage().setBody(inputMessages);

        final String createdJson
                = JsonUtils.convertMapToJson(operations.createStitchRequestBody(exchange.getMessage()).toMap());

        assertEquals(
                "{\"table_name\":\"table_1\",\"schema\":{\"field_1\":\"string\"},\"messages\":[{\"action\":\"upsert\",\"sequence\":1,\"data\":{\"field_1\":\"stitchMessage1\"}},"
                     + "{\"action\":\"upsert\",\"sequence\":2,\"data\":{\"field_1\":\"stitchMessage2-1\",\"field_2\":\"stitchMessage2-2\"}},{\"action\":\"upsert\",\"sequence\":3,\"data\":{\"field_1\":"
                     + "\"stitchMessage3\"}},{\"action\":\"upsert\",\"sequence\":4,\"data\":{\"field_1\":\"stitchMessage4\"}},{\"action\":\"upsert\",\"sequence\":5,\"data\":{\"field_1\":\"stitchMessage5\"}}],"
                     + "\"key_names\":[\"field_1\"]}",
                createdJson);
    }