Update Coordinator client, add datagram assembly.

logdog/client/cmd/logdog_cat/coordinatorSource.go

 // Copyright 2016 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 main
 
 import (
         "errors"
         "sync"
         "time"
@@ skipping 16 matching lines @@
 )
 
 // coordinatorSource is a fetcher.Source implementation that uses the
 // Coordiantor API.
 type coordinatorSource struct {
         sync.Mutex
 
         stream *coordinator.Stream
         tidx   types.MessageIndex
 
-        state coordinator.LogStream
+        streamState *coordinator.LogStream
 }
 
 func (s *coordinatorSource) LogEntries(c context.Context, req *fetcher.LogRequest) (
         []*logpb.LogEntry, types.MessageIndex, error) {
         s.Lock()
         defer s.Unlock()
 
-        p := coordinator.NewGetParams().Limit(int(req.Bytes), req.Count).Index(req.Index)
+        params := append(make([]coordinator.GetParam, 0, 4),
+                coordinator.LimitBytes(int(req.Bytes)),
+                coordinator.LimitCount(req.Count),
+                coordinator.Index(req.Index),
+        )
 
         // If we haven't terminated, use this opportunity to fetch/update our stream
         // state.
-        if s.tidx < 0 {
-                p = p.State(&s.state)
+        var streamState coordinator.LogStream
+        reqStream := (s.streamState == nil || s.streamState.State.TerminalIndex < 0)
+        if reqStream {
+                params = append(params, coordinator.WithState(&streamState))
         }
 
         for {
-                logs, err := s.stream.Get(c, p)
+                logs, err := s.stream.Get(c, params...)
                 switch err {
                 case nil:
-                        if s.state.State != nil && s.tidx < 0 {
-                                s.tidx = s.state.State.TerminalIndex
+                        if reqStream {
+                                s.streamState = &streamState
+                                s.tidx = streamState.State.TerminalIndex
                         }
                         return logs, s.tidx, nil
 
                 case coordinator.ErrNoSuchStream:
                         log.WithError(err).Warningf(c, "Stream does not exist. Sleeping pending registration.")
 
                         // Delay, interrupting if our Context is interrupted.
                         if tr := <-clock.After(c, noStreamDelay); tr.Incomplete() {
                                 return nil, 0, tr.Err
                         }
 
                 default:
                         if err != nil {
                                 return nil, 0, err
                         }
                 }
         }
 }
 
-func (s *coordinatorSource) getState() coordinator.LogStream {
-        s.Lock()
-        defer s.Unlock()
-        return s.state
+func (s *coordinatorSource) descriptor() (*logpb.LogStreamDescriptor, error) {
+        if s.streamState != nil {
+                return &s.streamState.Desc, nil
+        }
+        return nil, errors.New("no stream state loaded")
 }
-
-func (s *coordinatorSource) descriptor() (*logpb.LogStreamDescriptor, error) {
-        if d := s.state.Desc; d != nil {
-                return d, nil
-        }
-        return nil, errors.New("no descriptor loaded")
-}