gt:         TestUtil.setFieldValue(svc, "inFlightMsgIds", inFlightMsgIds);
context: /*******************************************************************************
   * Copyright (c) 2017, 2020 Eurotech and/or its affiliates and others
   * 
   * This program and the accompanying materials are made
   * available under the terms of the Eclipse Public License 2.0
   * which is available at https://www.eclipse.org/legal/epl-2.0/
   * 
   * SPDX-License-Identifier: EPL-2.0
   * 
   * Contributors:
   *  Eurotech
   ******************************************************************************/
  package org.eclipse.kura.core.data;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.fail;
  import static org.mockito.Matchers.anyInt;
  import static org.mockito.Matchers.anyObject;
  import static org.mockito.Matchers.eq;
  import static org.mockito.Mockito.doAnswer;
  import static org.mockito.Mockito.doThrow;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.ScheduledFuture;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.eclipse.kura.KuraConnectException;
  import org.eclipse.kura.KuraStoreException;
  import org.eclipse.kura.core.testutil.TestUtil;
  import org.eclipse.kura.data.DataTransportService;
  import org.eclipse.kura.data.DataTransportToken;
  import org.eclipse.kura.db.H2DbService;
  import org.eclipse.kura.status.CloudConnectionStatusEnum;
  import org.eclipse.kura.status.CloudConnectionStatusService;
  import org.eclipse.kura.watchdog.WatchdogService;
  import org.eclipse.paho.client.mqttv3.MqttException;
  import org.junit.Test;
  import org.osgi.service.component.ComponentContext;
  
  public class DataServiceImplTest {
  
      @Test
      public void testStartDbStore() throws Throwable {
          // test starting the store
  
          final int hkInterval = 1000;
          final int age = 100;
          final int capacity = 5000;
  
          DataServiceImpl svc = new DataServiceImpl();
  
          H2DbService dbServiceMock = mock(H2DbService.class);
  
          DataStore storeMock = mock(DataStore.class);
          TestUtil.setFieldValue(svc, "store", storeMock);
  
          Map<String, Object> properties = new HashMap<>();
          properties.put("store.housekeeper-interval", hkInterval);
          properties.put("store.purge-age", age);
          properties.put("store.capacity", capacity);
          DataServiceOptions dataServiceOptions = new DataServiceOptions(properties);
  
          TestUtil.setFieldValue(svc, "dataServiceOptions", dataServiceOptions);
  
          List<DataMessage> messages = new ArrayList<>();
          int id = 123;
          int pmi = 1234;
          String sessionId = "session";
          DataMessage msg = new DataMessage.Builder(id).withPublishedMessageId(pmi).withSessionId(sessionId).build();
          messages.add(msg);
          when(storeMock.allInFlightMessagesNoPayload()).thenReturn(messages);
  
          // the actual invocation
          svc.setH2DbService(dbServiceMock);
  
          verify(storeMock, times(1)).start(dbServiceMock, hkInterval, age, capacity);
  
          Map<DataTransportToken, Integer> ifMsgs = (Map<DataTransportToken, Integer>) TestUtil.getFieldValue(svc,
                  "inFlightMsgIds");
  
          assertEquals(1, ifMsgs.size());
          ifMsgs.forEach((key, value) -> {
              assertEquals(id, (int) value);
              assertEquals(pmi, key.getMessageId());
              assertEquals(sessionId, key.getSessionId());
          });
      }
  
      @Test
      public void testStartDbStoreErrorLog() throws Throwable {
          // produce the start store failure log message
  
          DataServiceImpl svc = new DataServiceImpl();
  
          H2DbService dbServiceMock = mock(H2DbService.class);
  
          DataStore storeMock = mock(DataStore.class);
          TestUtil.setFieldValue(svc, "store", storeMock);
  
          doThrow(new KuraStoreException("test")).when(storeMock).start(dbServiceMock, 900, 60, 10000);
  
          Map<String, Object> properties = new HashMap<>();
          DataServiceOptions dataServiceOptions = new DataServiceOptions(properties);
  
          TestUtil.setFieldValue(svc, "dataServiceOptions", dataServiceOptions);
  
          // the actual invocation
          svc.setH2DbService(dbServiceMock);
  
          verify(storeMock, times(1)).start(dbServiceMock, 900, 60, 10000);
      }
  
      @Test
      public void testConnectionEstablished() throws NoSuchFieldException, KuraStoreException {
          // new session, don't publish in-flight messages
  
          DataServiceImpl svc = new DataServiceImpl();
  
          CloudConnectionStatusService ccssMock = mock(CloudConnectionStatusService.class);
          svc.setCloudConnectionStatusService(ccssMock);
  
          DataStore storeMock = mock(DataStore.class);
          TestUtil.setFieldValue(svc, "store", storeMock);
  
          ComponentContext ctxMock = mock(ComponentContext.class);
          DataServiceListenerS dataServiceListeners = new DataServiceListenerS(ctxMock);
          TestUtil.setFieldValue(svc, "dataServiceListeners", dataServiceListeners);
  
          Map<String, Object> properties = new HashMap<>();
          properties.put("in-flight-messages.republish-on-new-session", false);
          DataServiceOptions dataServiceOptions = new DataServiceOptions(properties);
  
          TestUtil.setFieldValue(svc, "dataServiceOptions", dataServiceOptions);
  
          Map<DataTransportToken, Integer> inFlightMsgIds = mock(Map.class);
          TestUtil.setFieldValue(svc, "inFlightMsgIds", inFlightMsgIds);
  
          svc.onConnectionEstablished(true);
  
          verify(ccssMock, times(1)).updateStatus(svc, CloudConnectionStatusEnum.ON);
          verify(storeMock, times(1)).dropAllInFlightMessages();
          verify(inFlightMsgIds, times(1)).clear();
      }
  
      @Test
      public void testConnectionEstablishedErrorLog() throws NoSuchFieldException, KuraStoreException {
          // new session, don't publish in-flight messages, trigger error log
  
          DataServiceImpl svc = new DataServiceImpl();
  
          CloudConnectionStatusService ccssMock = mock(CloudConnectionStatusService.class);
          svc.setCloudConnectionStatusService(ccssMock);
  
          DataStore storeMock = mock(DataStore.class);
          TestUtil.setFieldValue(svc, "store", storeMock);
  
          doThrow(new KuraStoreException("test")).when(storeMock).dropAllInFlightMessages();
  
          ComponentContext ctxMock = mock(ComponentContext.class);
          DataServiceListenerS dataServiceListeners = new DataServiceListenerS(ctxMock);
          TestUtil.setFieldValue(svc, "dataServiceListeners", dataServiceListeners);
  
          Map<String, Object> properties = new HashMap<>();
          properties.put("in-flight-messages.republish-on-new-session", false);
          DataServiceOptions dataServiceOptions = new DataServiceOptions(properties);
  
          TestUtil.setFieldValue(svc, "dataServiceOptions", dataServiceOptions);
  
          Map<DataTransportToken, Integer> inFlightMsgIds = mock(Map.class);
          TestUtil.setFieldValue(svc, "inFlightMsgIds", inFlightMsgIds);
  
          svc.onConnectionEstablished(true);
  
          verify(ccssMock, times(1)).updateStatus(svc, CloudConnectionStatusEnum.ON);
          verify(storeMock, times(1)).dropAllInFlightMessages();
          verify(inFlightMsgIds, times(0)).clear();
      }
  
      @Test
      public void testConnectionEstablishedWithPublish() throws NoSuchFieldException, KuraStoreException {
          // new session, publish in-flight messages
  
          DataServiceImpl svc = new DataServiceImpl();
  
          CloudConnectionStatusService ccssMock = mock(CloudConnectionStatusService.class);
          svc.setCloudConnectionStatusService(ccssMock);
  
          DataStore storeMock = mock(DataStore.class);
          TestUtil.setFieldValue(svc, "store", storeMock);
  
          ComponentContext ctxMock = mock(ComponentContext.class);
          DataServiceListenerS dataServiceListeners = new DataServiceListenerS(ctxMock);
          TestUtil.setFieldValue(svc, "dataServiceListeners", dataServiceListeners);
  
          Map<String, Object> properties = new HashMap<>();
          properties.put("in-flight-messages.republish-on-new-session", true);
          DataServiceOptions dataServiceOptions = new DataServiceOptions(properties);
  
          TestUtil.setFieldValue(svc, "dataServiceOptions", dataServiceOptions);
  
          Map<DataTransportToken, Integer> inFlightMsgIds = mock(Map.class);
  