src/client/pfs.go
func (c APIClient) SubscribeCommitF(repo, branch, from string, state pfs.CommitState, f func(*pfs.CommitInfo) error) error {
	req := &pfs.SubscribeCommitRequest{
		Repo:   NewRepo(repo),
		Branch: branch,
		State:  state,
	}
	if from != "" {
		req.From = NewCommit(repo, from)
	}
	stream, err := c.PfsAPIClient.SubscribeCommit(c.Ctx(), req)
	if err != nil {
		return grpcutil.ScrubGRPC(err)
	}
	for {
		ci, err := stream.Recv()
		if err != nil {
			return grpcutil.ScrubGRPC(err)
		}
		if err := f(ci); err != nil {
			return grpcutil.ScrubGRPC(err)
		}
	}
}
func (c APIClient) SubscribeCommitF(repo, branch, from string, state pfs.CommitState, f func(*pfs.CommitInfo) error) error {
	req := &pfs.SubscribeCommitRequest{
		Repo:   NewRepo(repo),
		Branch: branch,
		State:  state,
	}
	if from != "" {
		req.From = NewCommit(repo, from)
	}
	stream, err := c.PfsAPIClient.SubscribeCommit(c.Ctx(), req)
	if err != nil {
		return grpcutil.ScrubGRPC(err)
	}
	for {
		ci, err := stream.Recv()
		if err != nil {
			return grpcutil.ScrubGRPC(err)
		}
		if err := f(ci); err != nil {
			return grpcutil.ScrubGRPC(err)
		}
	}
}
// SubscribeCommitF is like ListCommit but it calls a callback function with
// the results rather than returning an iterator.