func (s *Server) GetTask(ctx context.Context, in *tasks.GetTaskRequest) (*tasks.Task, error) {
	task, ok := s.fetchTask(in.GetName())
	if !ok {
		return nil, status.Errorf(codes.NotFound, "Task does not exist.")
	}
	if task == nil {
		return nil, status.Errorf(codes.FailedPrecondition, "The task no longer exists,  though a task with this name existed recently. The task either successfully completed or was deleted.")
	}
	return task.state, nil
}
GetTask returns the specified task
package main
import (
	"context"
	"flag"
	"fmt"
	"net"
	"regexp"
	"strings"
	"sync"
	tasks "google.golang.org/genproto/googleapis/cloud/tasks/v2"
	v1 "google.golang.org/genproto/googleapis/iam/v1"
	codes "google.golang.org/grpc/codes"
	status "google.golang.org/grpc/status"
	"github.com/golang/protobuf/proto"
	"github.com/golang/protobuf/ptypes/empty"
	"google.golang.org/grpc"
)
func NewServer() *Server {
	return &Server{
		qs: make(map[string]*Queue),
		ts: make(map[string]*Task),
		Options: ServerOptions{
			HardResetOnPurgeQueue: false,
		},
	}
}
type ServerOptions struct {
	HardResetOnPurgeQueue bool
}
type Server struct {
	qs map[string]*Queue
	ts map[string]*Task
	qsMux   sync.Mutex
	tsMux   sync.Mutex
	Options ServerOptions
}
func (s *Server) setQueue(queueName string, queue *Queue) {
	s.qsMux.Lock()
	defer s.qsMux.Unlock()
	s.qs[queueName] = queue
}
func (s *Server) fetchQueue(queueName string) (*Queue, bool) {
	s.qsMux.Lock()
	defer s.qsMux.Unlock()
	queue, ok := s.qs[queueName]
	return queue, ok
}
func (s *Server) removeQueue(queueName string) {
	s.setQueue(queueName, nil)
}
func (s *Server) setTask(taskName string, task *Task) {
	s.tsMux.Lock()
	defer s.tsMux.Unlock()
	s.ts[taskName] = task
}
func (s *Server) fetchTask(taskName string) (*Task, bool) {
	s.tsMux.Lock()
	defer s.tsMux.Unlock()
	task, ok := s.ts[taskName]
	return task, ok
}
func (s *Server) removeTask(taskName string) {
	s.setTask(taskName, nil)
}
func (s *Server) hardDeleteTask(taskName string) {
	s.tsMux.Lock()
	defer s.tsMux.Unlock()
	delete(s.ts, taskName)
}
func (s *Server) ListQueues(ctx context.Context, in *tasks.ListQueuesRequest) (*tasks.ListQueuesResponse, error) {
	var queueStates []*tasks.Queue
	s.qsMux.Lock()
	defer s.qsMux.Unlock()
	for _, queue := range s.qs {
		if queue != nil {
			queueStates = append(queueStates, queue.state)
		}
	}
	return &tasks.ListQueuesResponse{
		Queues: queueStates,
	}, nil
}
func (s *Server) GetQueue(ctx context.Context, in *tasks.GetQueueRequest) (*tasks.Queue, error) {
	queue, ok := s.fetchQueue(in.GetName())
	if !ok || queue == nil {
		return nil, status.Errorf(codes.NotFound, "Queue does not exist. If you just created the queue, wait at least a minute for the queue to initialize.")
	}
	return queue.state, nil
}
func (s *Server) CreateQueue(ctx context.Context, in *tasks.CreateQueueRequest) (*tasks.Queue, error) {
	queueState := in.GetQueue()
	name := queueState.GetName()
	nameMatched, _ := regexp.MatchString("projects/[A-Za-z0-9-]+/locations/[A-Za-z0-9-]+/queues/[A-Za-z0-9-]+", name)
	if !nameMatched {
		return nil, status.Errorf(codes.InvalidArgument, "Queue name must be formatted: \"projects/<PROJECT_ID>/locations/<LOCATION_ID>/queues/<QUEUE_ID>\"")
	}
	parent := in.GetParent()
	parentMatched, _ := regexp.MatchString("projects/[A-Za-z0-9-]+/locations/[A-Za-z0-9-]+", parent)
	if !parentMatched {
		return nil, status.Errorf(codes.InvalidArgument, "Invalid resource field value in the request.")
	}
	queue, ok := s.fetchQueue(name)
	if ok {
		if queue != nil {
			return nil, status.Errorf(codes.AlreadyExists, "Queue already exists")
		}
		return nil, status.Errorf(codes.FailedPrecondition, "The queue cannot be created because a queue with this name existed too recently.")
	}
	queue, queueState = NewQueue(
		name,
		proto.Clone(queueState).(*tasks.Queue),
		func(task *Task) {
			s.removeTask(task.state.GetName())
		},
	)
	s.setQueue(name, queue)
	queue.Run()
	return queueState, nil
}
func (s *Server) UpdateQueue(ctx context.Context, in *tasks.UpdateQueueRequest) (*tasks.Queue, error) {
	return nil, status.Errorf(codes.Unimplemented, "Not yet implemented")
}
func (s *Server) DeleteQueue(ctx context.Context, in *tasks.DeleteQueueRequest) (*empty.Empty, error) {
	queue, ok := s.fetchQueue(in.GetName())
	if !ok || queue == nil {
		return nil, status.Errorf(codes.NotFound, "Requested entity was not found.")
	}
	queue.Delete()
	s.removeQueue(in.GetName())
	return &empty.Empty{}, nil
}
func (s *Server) PurgeQueue(ctx context.Context, in *tasks.PurgeQueueRequest) (*tasks.Queue, error) {
	queue, _ := s.fetchQueue(in.GetName())
	if s.Options.HardResetOnPurgeQueue {
		queue.HardReset(s)
	} else {
		queue.Purge()
	}
	return queue.state, nil
}
func (s *Server) PauseQueue(ctx context.Context, in *tasks.PauseQueueRequest) (*tasks.Queue, error) {
	queue, _ := s.fetchQueue(in.GetName())
	queue.Pause()
	return queue.state, nil
}
func (s *Server) ResumeQueue(ctx context.Context, in *tasks.ResumeQueueRequest) (*tasks.Queue, error) {
	queue, _ := s.fetchQueue(in.GetName())
	queue.Resume()
	return queue.state, nil
}
func (s *Server) GetIamPolicy(ctx context.Context, in *v1.GetIamPolicyRequest) (*v1.Policy, error) {
	return nil, status.Errorf(codes.Unimplemented, "Not yet implemented")
}
func (s *Server) SetIamPolicy(ctx context.Context, in *v1.SetIamPolicyRequest) (*v1.Policy, error) {
	return nil, status.Errorf(codes.Unimplemented, "Not yet implemented")
}
func (s *Server) TestIamPermissions(ctx context.Context, in *v1.TestIamPermissionsRequest) (*v1.TestIamPermissionsResponse, error) {
	return nil, status.Errorf(codes.Unimplemented, "Not yet implemented")
}
func (s *Server) ListTasks(ctx context.Context, in *tasks.ListTasksRequest) (*tasks.ListTasksResponse, error) {
	queue, _ := s.fetchQueue(in.GetParent())
	var taskStates []*tasks.Task
	queue.tsMux.Lock()
	defer queue.tsMux.Unlock()
	for _, task := range queue.ts {
		if task != nil {
			taskStates = append(taskStates, task.state)
		}
	}
	return &tasks.ListTasksResponse{
		Tasks: taskStates,
	}, nil
}