LogDog: Update to archival V2.

appengine/logdog/coordinator/coordinatorTest/archival.go

Index: appengine/logdog/coordinator/coordinatorTest/archival.go
diff --git a/appengine/logdog/coordinator/coordinatorTest/archival.go b/appengine/logdog/coordinator/coordinatorTest/archival.go
new file mode 100644
index 0000000000000000000000000000000000000000..040933097db908672c81c5cbc91b33d63c1b4c9e
--- /dev/null
+++ b/appengine/logdog/coordinator/coordinatorTest/archival.go
@@ -0,0 +1,38 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package coordinatorTest
+
+import (
+	"fmt"
+	"sort"
+
+	"github.com/golang/protobuf/proto"
+	tq "github.com/luci/gae/service/taskqueue"
+	"github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
+	"github.com/luci/luci-go/common/logdog/types"
+)
+
+// GetArchiveTaskStreams returns the set of archive tasks in the supplied task
+// queue.
+//
+// If one of the tasks in the specified queue is not an ArchiveTask, an error
+// will be returned.
+func GetArchiveTaskStreams(ti tq.Interface, qname string) ([]string, error) {
+	queueContents := ti.Testable().GetScheduledTasks()[qname]
+
+	archiveTasks := make([]string, 0, len(queueContents))
+	for _, task := range queueContents {
+		var at logdog.ArchiveTask
+		if err := proto.Unmarshal(task.Payload, &at); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal payload: %v", err)
+		}
+
+		_, name := types.StreamPath(at.Path).Split()
+		archiveTasks = append(archiveTasks, string(name))
+	}
+
+	sort.Strings(archiveTasks)
+	return archiveTasks, nil
+}