replay - watcher.go

  package main
  
  import (
  	"log/slog"
  	"strings"
  	"sync"
  	"unsafe"
  
  	"github.com/fsnotify/fsnotify"
  )
  
  type watcher struct {
  	mu        sync.Mutex
  	listeners map[chan string]struct{}
  }
  
  func NewWatcher(base string) (*watcher, error) {
  	const xUnportableCloseWrite = 128
  
  	type withOpts struct {
  		bufsize    int
  		op         fsnotify.Op
  		sendCreate bool
  	}
  
  	type addOpt func(opt *withOpts)
  
  	fsw, err := fsnotify.NewWatcher()
  	if err != nil {
  		return nil, err
  	}
  
  	aw := (*fsnotify.Watcher).AddWith
  	awx := *(*func(*fsnotify.Watcher, string, ...addOpt) error)(unsafe.Pointer(&aw))
  
  	if err := awx(fsw, base, func(w *withOpts) { w.op = fsnotify.Rename | xUnportableCloseWrite }); err != nil {
  		return nil, err
  	}
  
  	w := &watcher{
  		listeners: make(map[chan string]struct{}),
  	}
  
  	go w.run(fsw)
  
  	return w, nil
  }
  
  func (w *watcher) run(fsw *fsnotify.Watcher) {
  	for {
  		select {
  		case ev := <-fsw.Events:
  			str := ev.String()
  			if strings.HasPrefix(str, "RENAME ") {
  				continue
  			} else if strings.HasPrefix(str, "CREATE ") {
  				str = "RENAME" + str[6:]
  			}
  
  			w.mu.Lock()
  			for ch := range w.listeners {
  				ch <- str
  			}
  			w.mu.Unlock()
  		case err := <-fsw.Errors:
  			slog.Error("fsnotify error", "err", err)
  		}
  	}
  }
  
  func (w *watcher) Listen() chan string {
  	ch := make(chan string)
  
  	w.mu.Lock()
  	w.listeners[ch] = struct{}{}
  	w.mu.Unlock()
  
  	return ch
  }
  
  func (w *watcher) Cancel(ch chan string) {
  	w.mu.Lock()
  	delete(w.listeners, ch)
  	w.mu.Unlock()
  }