Update Coordinator client, add datagram assembly.

logdog/client/coordinator/query_test.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package coordinator
 
 import (
         "errors"
         "fmt"
         "testing"
@@ skipping 83 matching lines @@
                                         "foo": "bar",
                                         "baz": "qux",
                                 },
                                 ContentType: "application/text",
                                 Before:      now,
                                 After:       now,
                                 Purged:      Both,
                                 State:       true,
                         }
 
+                        st := logdog.LogStreamState{
+                                Created: google.NewTimestamp(now),
+                        }
+
                         var results []*LogStream
                         accumulate := func(s *LogStream) bool {
                                 results = append(results, s)
                                 return true
                         }
 
                         Convey(`Can accumulate results across queries.`, func() {
                                 // This handler will return a single query per request, as well as a
                                 // non-empty Next pointer for the next query element. It progresses
                                 // "a" => "b" => "final" => "".
                                 svc.H = func(req *logdog.QueryRequest) (*logdog.QueryResponse, error) {
                                         r := logdog.QueryResponse{
                                                 Project: string(project),
                                         }
+
                                         switch req.Next {
                                         case "":
-                                                r.Streams = append(r.Streams, gen("a", nil))
+                                                r.Streams = append(r.Streams, gen("a", &st))
                                                 r.Next = "b"
                                         case "b":
-                                                r.Streams = append(r.Streams, gen("b", nil))
+                                                r.Streams = append(r.Streams, gen("b", &st))
                                                 r.Next = "final"
                                         case "final":
-                                                r.Streams = append(r.Streams, gen("final", nil))
+                                                r.Streams = append(r.Streams, gen("final", &st))
                                         default:
                                                 return nil, errors.New("invalid cursor")
                                         }
                                         return &r, nil
                                 }
 
                                 So(client.Query(c, project, path, q, accumulate), ShouldBeNil)
                                 So(results, shouldHaveLogStreams, "test/+/a", "test/+/b", "test/+/final")
                         })
 
                         Convey(`Will stop invoking the callback if it returns false.`, func() {
                                 // This handler will return three query results, "a", "b", and "c".
                                 svc.H = func(*logdog.QueryRequest) (*logdog.QueryResponse, error) {
                                         return &logdog.QueryResponse{
                                                 Streams: []*logdog.QueryResponse_Stream{
-                                                        gen("a", nil),
-                                                        gen("b", nil),
-                                                        gen("c", nil),
+                                                        gen("a", &st),
+                                                        gen("b", &st),
+                                                        gen("c", &st),
                                                 },
                                                 Next: "infiniteloop",
                                         }, nil
                                 }
 
                                 accumulate = func(s *LogStream) bool {
                                         results = append(results, s)
                                         return len(results) < 3
                                 }
                                 So(client.Query(c, project, path, q, accumulate), ShouldBeNil)
                                 So(results, shouldHaveLogStreams, "test/+/a", "test/+/b", "test/+/c")
                         })
 
                         Convey(`Will properly handle state and protobuf deserialization.`, func() {
                                 svc.H = func(*logdog.QueryRequest) (*logdog.QueryResponse, error) {
                                         return &logdog.QueryResponse{
                                                 Streams: []*logdog.QueryResponse_Stream{
                                                         gen("a", &logdog.LogStreamState{
                                                                 Created: google.NewTimestamp(now),
                                                         }),
                                                 },
                                         }, nil
                                 }
 
                                 So(client.Query(c, project, path, q, accumulate), ShouldBeNil)
                                 So(results, shouldHaveLogStreams, "test/+/a")
                                 So(results[0], ShouldResemble, &LogStream{
                                         Path: "test/+/a",
-                                        Desc: &logpb.LogStreamDescriptor{Prefix: "test", Name: "a"},
-                                        State: &StreamState{
+                                        Desc: logpb.LogStreamDescriptor{Prefix: "test", Name: "a"},
+                                        State: StreamState{
                                                 Created: now.UTC(),
                                         },
                                 })
                         })
 
                         Convey(`Can query for stream types`, func() {
                                 svc.H = func(*logdog.QueryRequest) (*logdog.QueryResponse, error) {
                                         return &logdog.QueryResponse{}, nil
                                 }
 
@@ skipping 27 matching lines @@
                         Convey(`Will return ErrNoAccess if permission denied.`, func() {
                                 svc.H = func(*logdog.QueryRequest) (*logdog.QueryResponse, error) {
                                         return nil, grpcutil.Unauthenticated
                                 }
 
                                 So(client.Query(c, project, path, q, accumulate), ShouldEqual, ErrNoAccess)
                         })
                 })
         })
 }