@Override
    public void process(Exchange exchange) throws Exception {

        if (logger.isDebugEnabled()) {
            logger.debug("uploader thread/id: {} / {}. api call completed.", Thread.currentThread().getId(),
                    exchange.getExchangeId());
        }

        if (exchange.getIn().getBody() instanceof List) {
            boolean groupedExchanges = false;
            for (Object body : exchange.getIn().getBody(List.class)) {
                if (body instanceof Exchange) {
                    send((Exchange) body);
                    groupedExchanges = true;
                }
            }
            if (!groupedExchanges) {
                send(exchange);
            }
        } else {
            send(exchange);
        }
    }
@Test
    public void testProcess() throws Exception {
        GooglePubsubLiteProducer producer = new GooglePubsubLiteProducer(mockEndpoint);
        String testPayload = "Test Payload";

        when(mockExchange.getIn()).thenReturn(mockMessage);
        when(mockEndpoint.getProjectId()).thenReturn(123456789012L);
        when(mockEndpoint.getLocation()).thenReturn("europe-west3");
        when(mockEndpoint.getDestinationName()).thenReturn("testDestination");
        when(mockEndpoint.getComponent()).thenReturn(mock(GooglePubsubLiteComponent.class));
        when(mockEndpoint.getComponent().getPublisher(any(), any())).thenReturn(mockPublisher);

        when(mockExchange.getMessage()).thenReturn(mockMessage);
        when(mockMessage.getBody()).thenReturn(testPayload.getBytes());
        when(mockExchange.getMessage().getHeader(GooglePubsubLiteConstants.ATTRIBUTES, Map.class)).thenReturn(null);
        when(mockExchange.getMessage().getHeader(GooglePubsubLiteConstants.ORDERING_KEY, String.class)).thenReturn(null);

        when(mockPublisher.publish(any())).thenReturn(ApiFutures.immediateFuture("messageId"));

        producer.process(mockExchange);

        verify(mockPublisher, times(1)).publish(any());
    }