Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.touchRelative(root, "in-foo-2")
# flush-subscriptions should let this come through immediately
flush = self.watchmanCommand(
"flush-subscriptions", root, {"sync_timeout": 1000}
)
del flush["version"]
self.assertDictEqual(
{"synced": ["defer"], "no_sync_needed": [], "dropped": []}, flush
)
sub_data = self.getSubscription("defer", root)
self.assertEqual(1, len(sub_data))
self.assertFileListsEqual(["in-foo-2"], sub_data[0]["files"])
self.touchRelative(root, "in-foo-3")
# We expect this to timeout because state=foo is asserted
with self.assertRaises(pywatchman.SocketTimeout):
self.waitForSub("defer", root, timeout=1)
self.watchmanCommand(
"state-leave", root, {"name": "foo", "metadata": "leavemeta"}
)
end = self.waitForSub(
"defer",
root,
accept=lambda x: self.matchStateSubscription(x, "state-leave"),
)[0]
self.assertEqual("leavemeta", end["metadata"])
# and now we should observe the file change
self.assertNotEqual(None, self.waitForSub("defer", root))
self.watchmanCommand("watch", root)
self.watchmanCommand(
"subscribe", root, "defer", {"fields": ["name"], "defer": ["foo"]}
)
self.touchRelative(root, "a")
self.assertNotEqual(None, self.waitForSub("defer", root=root))
self.watchmanCommand("state-enter", root, "foo")
begin = self.waitForSub("defer", root)[0]
self.assertEqual("foo", begin["state-enter"])
self.touchRelative(root, "in-foo")
# We expect this to timeout because state=foo is asserted
with self.assertRaises(pywatchman.SocketTimeout):
self.waitForSub("defer", root, timeout=1)
self.watchmanCommand("state-leave", root, "foo")
self.assertNotEqual(
None,
self.waitForSub(
"defer",
root,
accept=lambda x: self.matchStateSubscription(x, "state-leave"),
),
)
# and now we should observe the file change
self.assertNotEqual(None, self.waitForSub("defer", root))
query_params = format_watchman_query_params(
includes, excludes, include_dotfiles, relative_root, watchman_use_glob_generator
)
# Sync cookies cause a massive overhead when issuing thousands of
# glob queries. Only enable them (by not setting sync_timeout to 0)
# for the very first request issued by this process.
if sync_cookie_state.use_sync_cookies:
sync_cookie_state.use_sync_cookies = False
else:
query_params["sync_timeout"] = 0
query = ["query", watchman_watch_root, query_params]
try:
res = watchman_client.query(*query)
except pywatchman.SocketTimeout:
# If we just timed out, we might end up getting results eventually
# in a different query that were bound for this one. Call close
# here, and we'll reconnect if we try to query again
watchman_client.close()
# Watchman timeouts are not fatal. Fall back on the normal glob flow.
return None
error_message = res.get("error")
if error_message is not None:
diagnostics.append(
Diagnostic(
message=error_message, level="error", source="watchman", exception=None
)
)
warning_message = res.get("warning")
if warning_message is not None:
diagnostics.append(
cmd_buf = deque(command for command in reversed(commands))
self._connect()
while 1:
# Interleave sends and receives to avoid bi-directional communication issues.
if cmd_buf:
item = cmd_buf.pop()
try:
self.sendConn.send(item)
except pywatchman.SocketTimeout:
cmd_buf.append(item)
yield
try:
result = self.recvConn.receive()
except pywatchman.SocketTimeout:
# Socket timeout - yield runtime context.
yield
else:
if 'error' in result:
raise pywatchman.WatchmanError('error from watchman: {}'.format(result['error']))
elif self.isUnilateralResponse(result) or 'subscribe' in result:
yield result
else:
yield result
break
for change in changed:
if change in input_to_outputs:
result.input_to_outputs[change] = set(input_to_outputs[change])
else:
result.unrecognized.add(change)
for input, outputs in result.input_to_outputs.items():
for output in outputs:
if output not in result.output_to_inputs:
result.output_to_inputs[output] = set()
result.output_to_inputs[output].add(input)
yield result
except pywatchman.SocketTimeout:
# Let's check to see if we're still functional.
_version = self.client.query('version')
except pywatchman.CommandError as e:
# Abstract away pywatchman errors.
raise FasterBuildException(e, 'Command error using pywatchman to watch {}'.format(
self.config_environment.topsrcdir))
except pywatchman.SocketTimeout as e:
# Abstract away pywatchman errors.
raise FasterBuildException(e, 'Socket timeout using pywatchman to watch {}'.format(
self.config_environment.topsrcdir))
finally:
self.client.close()
# 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.
"""
# The CLI transport does not support pipelining.
if self.transport is pywatchman.CLIProcessTransport:
raise NotImplementedError()