def setup_watcher!(watch_type, opts)
      return yield unless opts.delete(:watch)

      @setup_watcher_mutex.synchronize do
        path = opts[:path]
        added, set = nil, nil

        synchronize do
          set = @outstanding_watches.fetch(watch_type)
          added = set.add?(path)
        end

        if added
          logger.debug { "adding watcher #{watch_type.inspect} for #{path.inspect}"}

          # if we added the path to the set, blocking further registration of
          # watches and an exception is raised then we rollback
          begin
            # this path has no outstanding watchers, let it do its thing
            opts[:watcher] = watcher_callback(watch_type)

            yield opts
          rescue Exception
            synchronize do
              set.delete(path)
            end
            raise
          end
        else
          logger.debug { "watcher #{watch_type.inspect} already set for #{path.inspect}"}

          # we did not add the path to the set, which means we are not
          # responsible for removing a block on further adds if the operation
          # fails, therefore, we just yield
          yield opts
        end
      end
    end
def setup_watcher!(watch_type, opts)
      return yield unless opts.delete(:watch)

      @setup_watcher_mutex.synchronize do
        path = opts[:path]
        added, set = nil, nil

        synchronize do
          set = @outstanding_watches.fetch(watch_type)
          added = set.add?(path)
        end

        if added
          logger.debug { "adding watcher #{watch_type.inspect} for #{path.inspect}"}

          # if we added the path to the set, blocking further registration of
          # watches and an exception is raised then we rollback
          begin
            # this path has no outstanding watchers, let it do its thing
            opts[:watcher] = watcher_callback(watch_type)

            yield opts
          rescue Exception
            synchronize do
              set.delete(path)
            end
            raise
          end
        else
          logger.debug { "watcher #{watch_type.inspect} already set for #{path.inspect}"}

          # we did not add the path to the set, which means we are not
          # responsible for removing a block on further adds if the operation
          # fails, therefore, we just yield
          yield opts
        end
      end
    end
implements not only setting up the watcher callback, but deduplicating
 event delivery. Keeps track of in-flight watcher-type+path requests and
 doesn't re-register the watcher with the server until a response has been
 fired. This prevents one event delivery to *every* callback per :watch => true
 argument.

 due to arguably poor design, we destructively modify opts before we yield
 and the client implictly knows this (this method constitutes some of the option
 parsing for the base class methods)

 @private