// MonitorJob returns a channel that can be passed into the deal monitoring process.
MonitorJob returns a channel that can be passed into the deal monitoring process.
// MonitorJob returns a channel that can be passed into the deal monitoring process .
func (s *Store) MonitorJob(j ffs.StorageJob) chan deals.StorageDealInfo {
	dealUpdates := make(chan deals.StorageDealInfo, 1000)
	go func() {
		for update := range dealUpdates {
			s.lock.Lock()
			_, ok := s.jobStatusCache[j.APIID]
			if !ok {
				s.jobStatusCache[j.APIID] = map[cid.Cid]map[cid.Cid]deals.StorageDealInfo{}
			}
			_, ok = s.jobStatusCache[j.APIID][j.Cid]
			if !ok {
				s.jobStatusCache[j.APIID][j.Cid] = map[cid.Cid]deals.StorageDealInfo{}
			}
			s.jobStatusCache[j.APIID][j.Cid][update.ProposalCid] = update
			job, err := s.get(j.ID)
			if err != nil {
				log.Errorf("getting job: %v", err)
				s.lock.Unlock()
				continue
			}
			values := make([]deals.StorageDealInfo, 0, len(s.jobStatusCache[j.APIID][j.Cid]))
			for _, v := range s.jobStatusCache[j.APIID][j.Cid] {
				values = append(values, v)
			}
			sort.Slice(values, func(i, j int) bool {
				return values[i].ProposalCid.String() < values[j].ProposalCid.String()
			})
			job.DealInfo = values
			if err := s.put(job); err != nil {
				log.Errorf("saving job with deal info updates: %v", err)
			}
			s.lock.Unlock()
		}
		s.lock.Lock()
		delete(s.jobStatusCache[j.APIID], j.Cid)
		s.lock.Unlock()
	}()
	return dealUpdates
}
func ( s * Store ) MonitorJob ( j ffs . StorageJob ) chan deals . StorageDealInfo { dealUpdates := make ( chan deals . StorageDealInfo , 1000 ) go func ( ) { for update := range dealUpdates { s . lock . Lock ( ) _ , ok := s . jobStatusCache [ j . APIID ] if ! ok { s . jobStatusCache [ j . APIID ] = map [ cid . Cid ] map [ cid . Cid ] deals . StorageDealInfo { } } _ , ok = s . jobStatusCache [ j . APIID ] [ j . Cid ] if ! ok { s . jobStatusCache [ j . APIID ] [ j . Cid ] = map [ cid . Cid ] deals . StorageDealInfo { } } s . jobStatusCache [ j . APIID ] [ j . Cid ] [ update . ProposalCid ] = update job , err := s . get ( j . ID ) if err != nil { log . Errorf ( " " , err ) s . lock . Unlock ( ) continue } values := make ( [ ] deals . StorageDealInfo , 0 , len ( s . jobStatusCache [ j . APIID ] [ j . Cid ] ) ) for _ , v := range s . jobStatusCache [ j . APIID ] [ j . Cid ] { values = append ( values , v ) } sort . Slice ( values , func ( i , j int ) bool { return values [ i ] . ProposalCid . String ( ) < values [ j ] . ProposalCid . String ( ) } ) job . DealInfo = values if err := s . put ( job ) ; err != nil { log . Errorf ( " " , err ) } s . lock . Unlock ( ) } s . lock . Lock ( ) delete ( s . jobStatusCache [ j . APIID ] , j . Cid ) s . lock . Unlock ( ) } ( ) return dealUpdates }
MonitorJob returns a channel that can be passed into the deal monitoring process.
MonitorJob returns a channel that can be passed into the deal monitoring process .