How to use the pywatchman.client function in pywatchman

self.proc = subprocess.Popen(
                args, env=env, stdin=None, stdout=cli_log_file, stderr=cli_log_file
            )
        if self.debug_watchman:
            print("Watchman instance PID: " + str(self.proc.pid))
            if pywatchman.compat.PYTHON3:
                user_input = input
            else:
                user_input = raw_input  # noqa:F821
            user_input("Press Enter to continue...")

        # wait for it to come up
        deadline = time.time() + self.start_timeout
        while time.time() < deadline:
            try:
                client = pywatchman.client(sockpath=self.sock_file)
                self.pid = client.query("get-pid")["pid"]
                break
            except pywatchman.SocketConnectError:
                t, val, tb = sys.exc_info()
                time.sleep(0.1)
            finally:
                client.close()

        if self.pid is None:
            # self.proc didn't come up: wait for it to die
            self.stop()
            pywatchman.compat.reraise(t, val, tb)

def getClient(self, inst=None, replace_cached=False, no_cache=False):
        if inst or not hasattr(self, "client") or no_cache:
            client = pywatchman.client(
                timeout=self.socketTimeout,
                transport=self.transport,
                sendEncoding=self.encoding,
                recvEncoding=self.encoding,
                sockpath=(inst or WatchmanInstance.getSharedInstance()).getSockPath(),
            )
            if (not inst or replace_cached) and not no_cache:
                # only cache the client if it points to the shared instance
                self.client = client
                self.addCleanup(lambda: self.__clearClient())
            return client
        return self.client

broken = True

    while not broken:
        test = tests_queue.get()
        try:
            if test == "terminate":
                break

            if Interrupt.wasInterrupted() or broken:
                continue

            result = None
            for attempt in range(0, args.retry_flaky + 1):
                # Check liveness of the server
                try:
                    client = pywatchman.client(timeout=3.0, sockpath=inst.getSockPath())
                    client.query("version")
                    client.close()
                except Exception as exc:
                    print(
                        "Failed to connect to watchman server: %s; starting a new one"
                        % exc
                    )

                    try:
                        inst.stop()
                    except Exception:
                        pass

                    try:
                        inst = WatchmanInstance.Instance(
                            {"watcher": args.watcher},

def test_client_property(self):
    self.assertIsInstance(self.watchman.client, pywatchman.client)

def input_changes(self, verbose=True):
        '''
        Return an iterator of `FasterBuildChange` instances as inputs
        to the faster build system change.
        '''

        # TODO: provide the debug diagnostics we want: this print is
        # not immediately before the watch.
        if verbose:
            print_line('watch', 'Connecting to watchman')
        # TODO: figure out why a large timeout is required for the
        # client, and a robust strategy for retrying timed out
        # requests.
        self.client = pywatchman.client(timeout=5.0)

        try:
            if verbose:
                print_line('watch', 'Checking watchman capabilities')
            # TODO: restrict these capabilities to the minimal set.
            self.client.capabilityCheck(required=[
                'clock-sync-timeout',
                'cmd-watch-project',
                'term-dirname',
                'wildmatch',
            ])

            if verbose:
                print_line('watch', 'Subscribing to {}'.format(self.config_environment.topsrcdir))
            self.subscribe_to_topsrcdir()
            if verbose:

watchman_client = None
    if options.use_watchman_glob:
        client_args = {"sendEncoding": "json", "recvEncoding": "json"}
        if options.watchman_query_timeout_ms is not None:
            # pywatchman expects a timeout as a nonnegative floating-point
            # value in seconds.
            client_args["timeout"] = max(
                0.0, options.watchman_query_timeout_ms / 1000.0
            )
        else:
            client_args["timeout"] = DEFAULT_WATCHMAN_QUERY_TIMEOUT
        if options.watchman_socket_path is not None:
            client_args["sockpath"] = options.watchman_socket_path
            client_args["transport"] = "local"
        watchman_client = pywatchman.client(**client_args)

    configs = {}
    if options.config is not None:
        with open(options.config, "rb") as f:
            for section, contents in iteritems(json.load(f)):
                for field, value in iteritems(contents):
                    configs[(section, field)] = value

    ignore_paths = []
    if options.ignore_paths is not None:
        with open(options.ignore_paths, "rb") as f:
            ignore_paths = [make_glob(i) for i in json.load(f)]

    build_file_processor = BuildFileProcessor(
        project_root,
        cell_roots,

# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from collections import deque

import pywatchman


# TODO(kwlzn): upstream this in pywatchman.
class StreamableWatchmanClient(pywatchman.client):
  """A watchman client subclass that provides for interruptable unilateral queries."""

  WatchmanError = pywatchman.WatchmanError
  SocketTimeout = pywatchman.SocketTimeout

  def stream_query(self, commands):
    """A generator of watchman events that allows queries to be pipelined and multiplexed. This
    continuously yields unilateral events and subscribe events, or None until an error condition
    or non-unilateral event (aside from subscribe) is received, at which point the generator
    ceases.

    The generator will yield None on socket timeouts unless the client's timeout has been set to
    None, in which case it will block indefinitely waiting on responses.

    :param iterable commands: An iterable of commands to send to watchman - e.g. one or more
                              subscribe commands.