func (pm *PrinterManager) printJob(nativePrinterName, filename, title, user, jobID string, ticket *cdd.CloudJobTicket, updateJob func(string, *cdd.PrintJobStateDiff) error) {
	defer os.Remove(filename)
	if !pm.addInFlightJob(jobID) {
		return
	}
	defer pm.deleteInFlightJob(jobID)
	if !pm.jobFullUsername {
		user = strings.Split(user, "@")[0]
	}
	printer, exists := pm.printers.GetByNativeName(nativePrinterName)
	if !exists {
		pm.incrementJobsProcessed(false)
		state := cdd.PrintJobStateDiff{
			State: &cdd.JobState{
				Type:               cdd.JobStateAborted,
				ServiceActionCause: &cdd.ServiceActionCause{ErrorCode: cdd.ServiceActionCausePrinterDeleted},
			},
		}
		if err := updateJob(jobID, &state); err != nil {
			log.ErrorJob(jobID, err)
		}
		return
	}
	nativeJobID, err := pm.native.Print(&printer, filename, title, user, jobID, ticket)
	if err != nil {
		pm.incrementJobsProcessed(false)
		log.ErrorJobf(jobID, "Failed to submit to native print system: %s", err)
		state := cdd.PrintJobStateDiff{
			State: &cdd.JobState{
				Type:              cdd.JobStateAborted,
				DeviceActionCause: &cdd.DeviceActionCause{ErrorCode: cdd.DeviceActionCausePrintFailure},
			},
		}
		if err := updateJob(jobID, &state); err != nil {
			log.ErrorJob(jobID, err)
		}
		return
	}
	log.InfoJobf(jobID, "Submitted as native job %d", nativeJobID)
	var state cdd.PrintJobStateDiff
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	defer pm.releaseJob(printer.Name, nativeJobID, jobID)
	for _ = range ticker.C {
		nativeState, err := pm.native.GetJobState(printer.Name, nativeJobID)
		if err != nil {
			log.WarningJobf(jobID, "Failed to get state of native job %d: %s", nativeJobID, err)
			state = cdd.PrintJobStateDiff{
				State: &cdd.JobState{
					Type:              cdd.JobStateAborted,
					DeviceActionCause: &cdd.DeviceActionCause{ErrorCode: cdd.DeviceActionCauseOther},
				},
				PagesPrinted: state.PagesPrinted,
			}
			if err := updateJob(jobID, &state); err != nil {
				log.ErrorJob(jobID, err)
			}
			pm.incrementJobsProcessed(false)
			return
		}
		if !reflect.DeepEqual(*nativeState, state) {
			state = *nativeState
			if err = updateJob(jobID, &state); err != nil {
				log.ErrorJob(jobID, err)
			}
			log.InfoJobf(jobID, "State: %s", state.State.Type)
		}
		if state.State.Type != cdd.JobStateInProgress && state.State.Type != cdd.JobStateStopped {
			if state.State.Type == cdd.JobStateDone {
				pm.incrementJobsProcessed(true)
			} else {
				pm.incrementJobsProcessed(false)
			}
			return
		}
	}
}
printJob prints a new job to a native printer, then polls the native job state
and updates the GCP/Privet job state. then returns when the job state is DONE
or ABORTED.

All errors are reported and logged from inside this function.
package manager
import (
	"fmt"
	"hash/adler32"
	"os"
	"reflect"
	"strings"
	"sync"
	"time"
	"github.com/google/cloud-print-connector/cdd"
	"github.com/google/cloud-print-connector/gcp"
	"github.com/google/cloud-print-connector/lib"
	"github.com/google/cloud-print-connector/log"
	"github.com/google/cloud-print-connector/notification"
	"github.com/google/cloud-print-connector/privet"
	"github.com/google/cloud-print-connector/xmpp"
)
type NativePrintSystem interface {
	GetPrinters() ([]lib.Printer, error)
	GetJobState(printerName string, jobID uint32) (*cdd.PrintJobStateDiff, error)
	Print(printer *lib.Printer, fileName, title, user, gcpJobID string, ticket *cdd.CloudJobTicket) (uint32, error)
	ReleaseJob(printerName string, jobID uint32) error
	RemoveCachedPPD(printerName string)
}
type PrinterManager struct {
	native NativePrintSystem
	gcp    *gcp.GoogleCloudPrint
	xmpp   *xmpp.XMPP
	privet *privet.Privet
	printers *lib.ConcurrentPrinterMap
	jobStatsMutex sync.Mutex
	jobsDone      uint
	jobsError     uint
	jobsInFlightMutex sync.Mutex
	jobsInFlight      map[string]struct{}
	nativeJobQueueSize uint
	jobFullUsername    bool
	shareScope         string
	quit   chan struct{}
	useFcm bool
}
func NewPrinterManager(native NativePrintSystem, gcp *gcp.GoogleCloudPrint, privet *privet.Privet, printerPollInterval time.Duration, nativeJobQueueSize uint, jobFullUsername bool, shareScope string, jobs <-chan *lib.Job, notifications <-chan notification.PrinterNotification, useFcm bool) (*PrinterManager, error) {
	var printers *lib.ConcurrentPrinterMap
	var queuedJobsCount map[string]uint
	var err error
	if gcp != nil {
		var gcpPrinters []lib.Printer
		gcpPrinters, queuedJobsCount, err = gcp.ListPrinters()
		if err != nil {
			return nil, err
		}
		for i := range gcpPrinters {
			gcpPrinters[i].NativeJobSemaphore = lib.NewSemaphore(nativeJobQueueSize)
		}
		printers = lib.NewConcurrentPrinterMap(gcpPrinters)
	} else {
		printers = lib.NewConcurrentPrinterMap(nil)
	}
	pm := PrinterManager{
		native: native,
		gcp:    gcp,
		privet: privet,
		printers: printers,
		jobStatsMutex: sync.Mutex{},
		jobsDone:      0,
		jobsError:     0,
		jobsInFlightMutex: sync.Mutex{},
		jobsInFlight:      make(map[string]struct{}),
		nativeJobQueueSize: nativeJobQueueSize,
		jobFullUsername:    jobFullUsername,
		shareScope:         shareScope,
		quit:   make(chan struct{}),
		useFcm: useFcm,
	}
	if err = pm.SyncPrinters(true); err != nil {
		return nil, err
	}
	if privet != nil {
		for _, printer := range pm.printers.GetAll() {
			err := privet.AddPrinter(printer, pm.printers.GetByNativeName)
			if err != nil {
				log.WarningPrinterf(printer.Name, "Failed to register locally: %s", err)
			} else {
				log.InfoPrinterf(printer.Name, "Registered locally")
			}
		}
	}
	pm.syncPrintersPeriodically(printerPollInterval)
	pm.listenNotifications(jobs, notifications)
	if gcp != nil {
		for gcpPrinterID := range queuedJobsCount {
			p, _ := printers.GetByGCPID(gcpPrinterID)
			go gcp.HandleJobs(&p, func() { pm.incrementJobsProcessed(false) })
		}
	}
	return &pm, nil
}
func (pm *PrinterManager) Quit() {
	close(pm.quit)
}
func (pm *PrinterManager) syncPrintersPeriodically(interval time.Duration) {
	go func() {
		t := time.NewTimer(interval)
		defer t.Stop()
		for {
			select {
			case <-t.C:
				if err := pm.SyncPrinters(false); err != nil {
					log.Error(err)
				}
				t.Reset(interval)
			case <-pm.quit:
				return
			}
		}
	}()
}
func (pm *PrinterManager) SyncPrinters(ignorePrivet bool) error {
	log.Debug("Synchronizing printers, stand by")
	nativePrinters, err := pm.native.GetPrinters()
	if err != nil {
		return fmt.Errorf("Sync failed while calling GetPrinters(): %s", err)
	}
	for i := range nativePrinters {
		h := adler32.New()
		lib.DeepHash(nativePrinters[i].Tags, h)
		nativePrinters[i].Tags["tagshash"] = fmt.Sprintf("%x", h.Sum(nil))
		h = adler32.New()
		lib.DeepHash(nativePrinters[i].Description, h)
		nativePrinters[i].CapsHash = fmt.Sprintf("%x", h.Sum(nil))
		if pm.useFcm {
			nativePrinters[i].NotificationChannel = gcp.FCP_CHANNEL
		} else {
			nativePrinters[i].NotificationChannel = gcp.XMPP_CHANNEL
		}
	}
	diffs := lib.DiffPrinters(nativePrinters, pm.printers.GetAll())
	if diffs == nil {
		log.Debugf("Printers are already in sync; there are %d", len(nativePrinters))
		return nil
	}
	ch := make(chan lib.Printer, len(diffs))
	for i := range diffs {
		go pm.applyDiff(&diffs[i], ch, ignorePrivet)
	}
	currentPrinters := make([]lib.Printer, 0, len(diffs))
	for _ = range diffs {
		p := <-ch
		if p.Name != "" {
			currentPrinters = append(currentPrinters, p)
		}
	}
	pm.printers.Refresh(currentPrinters)
	log.Debugf("Finished synchronizing %d printers", len(currentPrinters))
	return nil
}
func (pm *PrinterManager) applyDiff(diff *lib.PrinterDiff, ch chan<- lib.Printer, ignorePrivet bool) {
	switch diff.Operation {
	case lib.RegisterPrinter:
		if pm.gcp != nil {
			if err := pm.gcp.Register(&diff.Printer); err != nil {
				log.ErrorPrinterf(diff.Printer.Name, "Failed to register: %s", err)
				break
			}
			log.InfoPrinterf(diff.Printer.Name+" "+diff.Printer.GCPID, "Registered in the cloud")
			if pm.gcp.CanShare() {
				if err := pm.gcp.Share(diff.Printer.GCPID, pm.shareScope, gcp.User, true, false); err != nil {
					log.ErrorPrinterf(diff.Printer.Name, "Failed to share: %s", err)
				} else {
					log.InfoPrinterf(diff.Printer.Name, "Shared")
				}
			}
		}
		diff.Printer.NativeJobSemaphore = lib.NewSemaphore(pm.nativeJobQueueSize)
		if pm.privet != nil && !ignorePrivet {
			err := pm.privet.AddPrinter(diff.Printer, pm.printers.GetByNativeName)
			if err != nil {
				log.WarningPrinterf(diff.Printer.Name, "Failed to register locally: %s", err)
			} else {
				log.InfoPrinterf(diff.Printer.Name, "Registered locally")
			}
		}
		ch <- diff.Printer
		return
	case lib.UpdatePrinter:
		if pm.gcp != nil {
			if err := pm.gcp.Update(diff); err != nil {
				log.ErrorPrinterf(diff.Printer.Name+" "+diff.Printer.GCPID, "Failed to update: %s", err)
			} else {
				log.InfoPrinterf(diff.Printer.Name+" "+diff.Printer.GCPID, "Updated in the cloud")
			}
		}
		if pm.privet != nil && !ignorePrivet && diff.DefaultDisplayNameChanged {
			err := pm.privet.UpdatePrinter(diff)
			if err != nil {
				log.WarningPrinterf(diff.Printer.Name, "Failed to update locally: %s", err)
			} else {
				log.InfoPrinterf(diff.Printer.Name, "Updated locally")
			}
		}
		ch <- diff.Printer
		return
	case lib.DeletePrinter:
		pm.native.RemoveCachedPPD(diff.Printer.Name)
		if pm.gcp != nil {
			if err := pm.gcp.Delete(diff.Printer.GCPID); err != nil {
				log.ErrorPrinterf(diff.Printer.Name+" "+diff.Printer.GCPID, "Failed to delete from the cloud: %s", err)
				break
			}
			log.InfoPrinterf(diff.Printer.Name+" "+diff.Printer.GCPID, "Deleted from the cloud")
		}
		if pm.privet != nil && !ignorePrivet {
			err := pm.privet.DeletePrinter(diff.Printer.Name)
			if err != nil {
				log.WarningPrinterf(diff.Printer.Name, "Failed to delete: %s", err)
			} else {
				log.InfoPrinterf(diff.Printer.Name, "Deleted locally")
			}
		}
	case lib.NoChangeToPrinter:
		ch <- diff.Printer
		return
	}
	ch <- lib.Printer{}
}
func (pm *PrinterManager) listenNotifications(jobs <-chan *lib.Job, messages <-chan notification.PrinterNotification) {
	go func() {
		for {
			select {
			case <-pm.quit:
				return
			case job := <-jobs:
				log.DebugJobf(job.JobID, "Received job: %+v", job)
				go pm.printJob(job.NativePrinterName, job.Filename, job.Title, job.User, job.JobID, job.Ticket, job.UpdateJob)
			case message := <-messages:
				log.Debugf("Received message: %+v", message)
				if message.Type == notification.PrinterNewJobs {
					if p, exists := pm.printers.GetByGCPID(message.GCPID); exists {
						go pm.gcp.HandleJobs(&p, func() { pm.incrementJobsProcessed(false) })
					}
				}
			}
		}
	}()
}
func (pm *PrinterManager) incrementJobsProcessed(success bool) {
	pm.jobStatsMutex.Lock()
	defer pm.jobStatsMutex.Unlock()
	if success {
		pm.jobsDone += 1
	} else {
		pm.jobsError += 1
	}
}
func (pm *PrinterManager) addInFlightJob(jobID string) bool {
	pm.jobsInFlightMutex.Lock()
	defer pm.jobsInFlightMutex.Unlock()
	if _, exists := pm.jobsInFlight[jobID]; exists {
		return false
	}
	pm.jobsInFlight[jobID] = struct{}{}
	return true
}
func (pm *PrinterManager) deleteInFlightJob(jobID string) {
	pm.jobsInFlightMutex.Lock()
	defer pm.jobsInFlightMutex.Unlock()
	delete(pm.jobsInFlight, jobID)
}