waiter.go

Documentation: github.com/rs/zerolog/diode/internal/diodes

     1  package diodes
     2  
     3  import (
     4  	"context"
     5  	"sync"
     6  )
     7  
     8  // Waiter will use a conditional mutex to alert the reader to when data is
     9  // available.
    10  type Waiter struct {
    11  	Diode
    12  	mu  sync.Mutex
    13  	c   *sync.Cond
    14  	ctx context.Context
    15  }
    16  
    17  // WaiterConfigOption can be used to setup the waiter.
    18  type WaiterConfigOption func(*Waiter)
    19  
    20  // WithWaiterContext sets the context to cancel any retrieval (Next()). It
    21  // will not change any results for adding data (Set()). Default is
    22  // context.Background().
    23  func WithWaiterContext(ctx context.Context) WaiterConfigOption {
    24  	return func(c *Waiter) {
    25  		c.ctx = ctx
    26  	}
    27  }
    28  
    29  // NewWaiter returns a new Waiter that wraps the given diode.
    30  func NewWaiter(d Diode, opts ...WaiterConfigOption) *Waiter {
    31  	w := new(Waiter)
    32  	w.Diode = d
    33  	w.c = sync.NewCond(&w.mu)
    34  	w.ctx = context.Background()
    35  
    36  	for _, opt := range opts {
    37  		opt(w)
    38  	}
    39  
    40  	go func() {
    41  		<-w.ctx.Done()
    42  		w.c.Broadcast()
    43  	}()
    44  
    45  	return w
    46  }
    47  
    48  // Set invokes the wrapped diode's Set with the given data and uses Broadcast
    49  // to wake up any readers.
    50  func (w *Waiter) Set(data GenericDataType) {
    51  	w.Diode.Set(data)
    52  	w.c.Broadcast()
    53  }
    54  
    55  // Next returns the next data point on the wrapped diode. If there is not any
    56  // new data, it will Wait for set to be called or the context to be done.
    57  // If the context is done, then nil will be returned.
    58  func (w *Waiter) Next() GenericDataType {
    59  	w.mu.Lock()
    60  	defer w.mu.Unlock()
    61  
    62  	for {
    63  		data, ok := w.Diode.TryNext()
    64  		if !ok {
    65  			if w.isDone() {
    66  				return nil
    67  			}
    68  
    69  			w.c.Wait()
    70  			continue
    71  		}
    72  		return data
    73  	}
    74  }
    75  
    76  func (w *Waiter) isDone() bool {
    77  	select {
    78  	case <-w.ctx.Done():
    79  		return true
    80  	default:
    81  		return false
    82  	}
    83  }
    84
View as plain text