/*
monitor tracks changes in the specified subdirectory and sends out notifications
to the caller. This uses callbacks over channels to save on threads.
*/
monitor tracks changes in the specified subdirectory and sends out notifications
to the caller. This uses callbacks over channels to save on threads.
/ * monitor tracks changes in the specified subdirectory and sends out notifications to the caller . This uses callbacks over channels to save on threads . * /
func (r *exportedServiceMonitor) monitor() {
	var resp *etcd.GetResponse
	var ch etcd.WatchChan
	var wr etcd.WatchResponse
	var val *mvccpb.KeyValue
	var err error
	if resp, err = r.etcdKV.Get(context.Background(), r.basePath,
		etcd.WithPrefix()); err != nil {
		r.receiver.ReportError(err)
	}
	if resp != nil {
		for _, val = range resp.Kvs {
			var service = new(ExportedServiceRecord)
			var notification ExportedServiceUpdateNotification
			err = proto.Unmarshal(val.Value, service)
			if err != nil {
				r.receiver.ReportError(err)
				continue
			}
			notification = ExportedServiceUpdateNotification{
				Path:        string(val.Key),
				Update:      ExportedServiceUpdateNotification_NEW,
				UpdatedData: service,
			}
			r.receiver.ReportChange(&notification)
		}
	}
	ch = r.etcdWatcher.Watch(context.Background(), r.basePath,
		etcd.WithPrefix(), etcd.WithPrevKV())
	for wr = range ch {
		var ev *etcd.Event
		if wr.Err() != nil && !wr.Canceled {
			r.receiver.ReportError(wr.Err())
			continue
		}
		if wr.Canceled {
			err = wr.Err()
			if err == nil {
				r.receiver.ReportCancelled()
			} else {
				r.receiver.ReportFatal(err)
			}
			return
		}
		for _, ev = range wr.Events {
			if ev.IsCreate() {
				var service = new(ExportedServiceRecord)
				var notification ExportedServiceUpdateNotification
				val = ev.Kv
				err = proto.Unmarshal(val.Value, service)
				if err != nil {
					r.receiver.ReportError(err)
					continue
				}
				notification = ExportedServiceUpdateNotification{
					Path:        string(val.Key),
					Update:      ExportedServiceUpdateNotification_NEW,
					UpdatedData: service,
				}
				r.receiver.ReportChange(&notification)
			} else if ev.Type == mvccpb.DELETE {
				var service = new(ExportedServiceRecord)
				val = ev.PrevKv
				err = proto.Unmarshal(val.Value, service)
				if err != nil {
					r.receiver.ReportError(err)
					continue
				}
				var notification = ExportedServiceUpdateNotification{
					Path:        string(val.Key),
					Update:      ExportedServiceUpdateNotification_DELETED,
					UpdatedData: service,
				}
				r.receiver.ReportChange(&notification)
			} else {
				r.receiver.ReportError(fmt.Errorf(
					"Caught event I have no idea about: %v", ev))
			}
		}
	}
}
func ( r * exportedServiceMonitor ) monitor ( ) { var resp * etcd . GetResponse var ch etcd . WatchChan var wr etcd . WatchResponse var val * mvccpb . KeyValue var err error if resp , err = r . etcdKV . Get ( context . Background ( ) , r . basePath , etcd . WithPrefix ( ) ) ; err != nil { r . receiver . ReportError ( err ) } if resp != nil { for _ , val = range resp . Kvs { var service = new ( ExportedServiceRecord ) var notification ExportedServiceUpdateNotification err = proto . Unmarshal ( val . Value , service ) if err != nil { r . receiver . ReportError ( err ) continue } notification = ExportedServiceUpdateNotification { Path : string ( val . Key ) , Update : ExportedServiceUpdateNotification_NEW , UpdatedData : service , } r . receiver . ReportChange ( & notification ) } } ch = r . etcdWatcher . Watch ( context . Background ( ) , r . basePath , etcd . WithPrefix ( ) , etcd . WithPrevKV ( ) ) for wr = range ch { var ev * etcd . Event if wr . Err ( ) != nil && ! wr . Canceled { r . receiver . ReportError ( wr . Err ( ) ) continue } if wr . Canceled { err = wr . Err ( ) if err == nil { r . receiver . ReportCancelled ( ) } else { r . receiver . ReportFatal ( err ) } return } for _ , ev = range wr . Events { if ev . IsCreate ( ) { var service = new ( ExportedServiceRecord ) var notification ExportedServiceUpdateNotification val = ev . Kv err = proto . Unmarshal ( val . Value , service ) if err != nil { r . receiver . ReportError ( err ) continue } notification = ExportedServiceUpdateNotification { Path : string ( val . Key ) , Update : ExportedServiceUpdateNotification_NEW , UpdatedData : service , } r . receiver . ReportChange ( & notification ) } else if ev . Type == mvccpb . DELETE { var service = new ( ExportedServiceRecord ) val = ev . PrevKv err = proto . Unmarshal ( val . Value , service ) if err != nil { r . receiver . ReportError ( err ) continue } var notification = ExportedServiceUpdateNotification { Path : string ( val . Key ) , Update : ExportedServiceUpdateNotification_DELETED , UpdatedData : service , } r . receiver . ReportChange ( & notification ) } else { r . receiver . ReportError ( fmt . Errorf ( " " , ev ) ) } } } }
monitor tracks changes in the specified subdirectory and sends out notifications
to the caller.
monitor tracks changes in the specified subdirectory and sends out notifications to the caller .
/*
	   First, determine the current state of all objects in the tree.
	*/ /*
	   Start watching for changes on the specified prefix.
	*/ // Why is there no IsDelete()?!