Schedule active context jobs upon initialization(b/139435527).

Author: vkryachko

transport/transport-runtime/src/main/java/com/google/android/datatransport/runtime/TransportRuntime.java

@@ -20,6 +20,7 @@
 import com.google.android.datatransport.TransportFactory;
 import com.google.android.datatransport.runtime.scheduling.Scheduler;
 import com.google.android.datatransport.runtime.scheduling.jobscheduling.Uploader;
+import com.google.android.datatransport.runtime.scheduling.jobscheduling.WorkInitializer;
 import com.google.android.datatransport.runtime.time.Clock;
 import com.google.android.datatransport.runtime.time.Monotonic;
 import com.google.android.datatransport.runtime.time.WallTime;
@@ -48,11 +49,14 @@ public class TransportRuntime implements TransportInternal {
       @WallTime Clock eventClock,
       @Monotonic Clock uptimeClock,
       Scheduler scheduler,
-      Uploader uploader) {
+      Uploader uploader,
+      WorkInitializer initializer) {
     this.eventClock = eventClock;
     this.uptimeClock = uptimeClock;
     this.scheduler = scheduler;
     this.uploader = uploader;
+
+    initializer.ensureContextsScheduled();
   }
 
   /**

transport/transport-runtime/src/main/java/com/google/android/datatransport/runtime/scheduling/jobscheduling/WorkInitializer.java

@@ -0,0 +1,49 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.android.datatransport.runtime.scheduling.jobscheduling;
+
+import com.google.android.datatransport.runtime.TransportContext;
+import com.google.android.datatransport.runtime.scheduling.persistence.EventStore;
+import com.google.android.datatransport.runtime.synchronization.SynchronizationGuard;
+import java.util.concurrent.Executor;
+import javax.inject.Inject;
+
+public class WorkInitializer {
+  private final Executor executor;
+  private final EventStore store;
+  private final WorkScheduler scheduler;
+  private final SynchronizationGuard guard;
+
+  @Inject
+  WorkInitializer(
+      Executor executor, EventStore store, WorkScheduler scheduler, SynchronizationGuard guard) {
+    this.executor = executor;
+    this.store = store;
+    this.scheduler = scheduler;
+    this.guard = guard;
+  }
+
+  public void ensureContextsScheduled() {
+    executor.execute(
+        () ->
+            guard.runCriticalSection(
+                () -> {
+                  for (TransportContext context : store.loadActiveContexts()) {
+                    scheduler.schedule(context, 1);
+                  }
+                  return null;
+                }));
+  }
+}

transport/transport-runtime/src/main/java/com/google/android/datatransport/runtime/scheduling/persistence/EventStore.java

@@ -50,6 +50,8 @@ public interface EventStore extends Closeable {
   /** Load all pending events for a given backend. */
   Iterable<PersistedEvent> loadBatch(TransportContext transportContext);
 
+  Iterable<TransportContext> loadActiveContexts();
+
   /** Remove events that have been stored for more than 7 days. */
   int cleanUp();
 }

transport/transport-runtime/src/main/java/com/google/android/datatransport/runtime/scheduling/persistence/SQLiteEventStore.java

@@ -274,6 +274,28 @@ public Iterable<PersistedEvent> loadBatch(TransportContext transportContext) {
         });
   }
 
+  @Override
+  public Iterable<TransportContext> loadActiveContexts() {
+    return inTransaction(
+        db ->
+            tryWithCursor(
+                db.rawQuery(
+                    "SELECT t.backend_name, t.priority, t.extras FROM transport_contexts AS t, events AS e WHERE e.context_id = t._id",
+                    new String[] {}),
+                cursor -> {
+                  List<TransportContext> results = new ArrayList<>();
+                  while (cursor.moveToNext()) {
+                    results.add(
+                        TransportContext.builder()
+                            .setBackendName(cursor.getString(0))
+                            .setPriority(cursor.getInt(1))
+                            .setExtras(maybeBase64Decode(cursor.getString(2)))
+                            .build());
+                  }
+                  return results;
+                }));
+  }
+
   @Override
   public int cleanUp() {
     long oneWeekAgo = wallClock.getTime() - config.getEventCleanUpAge();
@@ -286,6 +308,13 @@ public void close() {
     schemaManager.close();
   }
 
+  private static byte[] maybeBase64Decode(@Nullable String value) {
+    if (value == null) {
+      return null;
+    }
+    return Base64.decode(value, Base64.DEFAULT);
+  }
+
   /** Loads all events for a backend. */
   private List<PersistedEvent> loadEvents(SQLiteDatabase db, TransportContext transportContext) {
     List<PersistedEvent> events = new ArrayList<>();

transport/transport-runtime/src/test/java/com/google/android/datatransport/runtime/TransportRuntimeTest.java

@@ -30,6 +30,7 @@
 import com.google.android.datatransport.runtime.backends.TransportBackend;
 import com.google.android.datatransport.runtime.scheduling.ImmediateScheduler;
 import com.google.android.datatransport.runtime.scheduling.jobscheduling.Uploader;
+import com.google.android.datatransport.runtime.scheduling.jobscheduling.WorkInitializer;
 import java.util.Collections;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -40,9 +41,11 @@
 public class TransportRuntimeTest {
   private static final String TEST_KEY = "test";
   private static final String TEST_VALUE = "test-value";
-  private TransportInternal transportInternalMock = mock(TransportInternal.class);
-  private TransportBackend mockBackend = mock(TransportBackend.class);
-  private BackendRegistry mockRegistry = mock(BackendRegistry.class);
+
+  private final TransportInternal transportInternalMock = mock(TransportInternal.class);
+  private final TransportBackend mockBackend = mock(TransportBackend.class);
+  private final BackendRegistry mockRegistry = mock(BackendRegistry.class);
+  private final WorkInitializer mockInitializer = mock(WorkInitializer.class);
 
   @Test
   public void testTransportInternalSend() {
@@ -75,7 +78,10 @@ public void testTransportRuntimeBackendDiscovery() {
             () -> eventMillis,
             () -> uptimeMillis,
             new ImmediateScheduler(Runnable::run, mockRegistry),
-            new Uploader(null, null, null, null, null, null, () -> 2));
+            new Uploader(null, null, null, null, null, null, () -> 2),
+            mockInitializer);
+    verify(mockInitializer, times(1)).ensureContextsScheduled();
+
     when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
     when(mockBackend.decorate(any()))
         .thenAnswer(

transport/transport-runtime/src/test/java/com/google/android/datatransport/runtime/scheduling/jobscheduling/WorkInitializerTest.java

@@ -0,0 +1,53 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.android.datatransport.runtime.scheduling.jobscheduling;
+
+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 com.google.android.datatransport.runtime.TransportContext;
+import com.google.android.datatransport.runtime.scheduling.persistence.EventStore;
+import com.google.android.datatransport.runtime.synchronization.SynchronizationGuard.CriticalSection;
+import java.util.Arrays;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class WorkInitializerTest {
+
+  private final EventStore mockStore = mock(EventStore.class);
+  private final WorkScheduler mockScheduler = mock(WorkScheduler.class);
+
+  private final WorkInitializer initializer =
+      new WorkInitializer(Runnable::run, mockStore, mockScheduler, CriticalSection::execute);
+
+  @Test
+  public void test() {
+    TransportContext ctx1 =
+        TransportContext.builder().setBackendName("backend1").setExtras(null).build();
+    TransportContext ctx2 =
+        TransportContext.builder().setBackendName("backend1").setExtras("e1".getBytes()).build();
+
+    when(mockStore.loadActiveContexts()).thenReturn(Arrays.asList(ctx1, ctx2));
+
+    initializer.ensureContextsScheduled();
+
+    verify(mockScheduler, times(1)).schedule(ctx1, 1);
+    verify(mockScheduler, times(1)).schedule(ctx2, 1);
+  }
+}

transport/transport-runtime/src/test/java/com/google/android/datatransport/runtime/scheduling/persistence/InMemoryEventStore.java

@@ -102,6 +102,18 @@ public synchronized Iterable<PersistedEvent> loadBatch(TransportContext transpor
     return events;
   }
 
+  @Override
+  public synchronized Iterable<TransportContext> loadActiveContexts() {
+    List<TransportContext> results = new ArrayList<>();
+    for (Map.Entry<TransportContext, Map<Long, EventInternal>> entry : store.entrySet()) {
+      if (entry.getValue().isEmpty()) {
+        continue;
+      }
+      results.add(entry.getKey());
+    }
+    return results;
+  }
+
   @Override
   public int cleanUp() {
     return 0;

transport/transport-runtime/src/test/java/com/google/android/datatransport/runtime/scheduling/persistence/SQLiteEventStoreTest.java

@@ -268,4 +268,36 @@ public void cleanUp_whenEventIsOld_shouldDeleteIt() {
     assertThat(store.cleanUp()).isEqualTo(1);
     assertThat(store.loadBatch(TRANSPORT_CONTEXT)).isEmpty();
   }
+
+  @Test
+  public void loadActiveContexts_whenNoContextsAvailable_shouldReturnEmptyList() {
+    assertThat(store.loadActiveContexts()).isEmpty();
+  }
+
+  @Test
+  public void loadActiveContexts_whenTwoContextsAvailable_shouldReturnThem() {
+    TransportContext ctx1 =
+        TransportContext.builder().setBackendName("backend1").setExtras(null).build();
+    TransportContext ctx2 =
+        TransportContext.builder().setBackendName("backend1").setExtras("e1".getBytes()).build();
+
+    store.persist(ctx1, EVENT);
+    store.persist(ctx2, EVENT);
+
+    assertThat(store.loadActiveContexts()).containsExactly(ctx1, ctx2);
+  }
+
+  @Test
+  public void loadActiveContexts_whenTwoContextsWithOneAvailable_shouldReturnIt() {
+    TransportContext ctx1 =
+        TransportContext.builder().setBackendName("backend1").setExtras(null).build();
+    TransportContext ctx2 =
+        TransportContext.builder().setBackendName("backend1").setExtras("e1".getBytes()).build();
+
+    store.persist(ctx1, EVENT);
+    PersistedEvent persistedEvent2 = store.persist(ctx2, EVENT);
+    store.recordSuccess(Collections.singleton(persistedEvent2));
+
+    assertThat(store.loadActiveContexts()).containsExactly(ctx1);
+  }
 }