<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import functools <EOL> import json <EOL> import logging <EOL> import os <EOL> from collections import namedtuple <EOL> from pants . pantsd . process_manager import ProcessManager <EOL> from pants . pantsd . watchman_client import StreamableWatchmanClient <EOL> from pants . util . dirutil import safe_file_dump , safe_mkdir <EOL> from pants . util . retry import retry_on_exception <EOL> class Watchman ( ProcessManager ) : <EOL> """<STR_LIT>""" <EOL> SOCKET_TIMEOUT_SECONDS = <NUM_LIT> <EOL> EventHandler = namedtuple ( '<STR_LIT>' , [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def __init__ ( self , watchman_path , work_dir , log_level = '<STR_LIT:1>' , timeout = SOCKET_TIMEOUT_SECONDS ) : <EOL> """<STR_LIT>""" <EOL> super ( Watchman , self ) . __init__ ( name = '<STR_LIT>' , process_name = '<STR_LIT>' , socket_type = str ) <EOL> self . _watchman_path = self . _normalize_watchman_path ( watchman_path ) <EOL> self . _watchman_work_dir = os . path . join ( work_dir , self . name ) <EOL> self . _log_level = log_level <EOL> self . _timeout = timeout <EOL> self . _state_file = os . path . join ( self . _watchman_work_dir , '<STR_LIT>' . format ( self . name ) ) <EOL> self . _log_file = os . path . join ( self . _watchman_work_dir , '<STR_LIT>' . format ( self . name ) ) <EOL> self . _sock_file = os . path . join ( self . _watchman_work_dir , '<STR_LIT>' . format ( self . name ) ) <EOL> self . _logger = logging . getLogger ( __name__ ) <EOL> self . _watchman_client = None <EOL> @ property <EOL> def client ( self ) : <EOL> if not self . _watchman_client : <EOL> self . _watchman_client = self . _make_client ( ) <EOL> return self . _watchman_client <EOL> def _make_client ( self ) : <EOL> """<STR_LIT>""" <EOL> return StreamableWatchmanClient ( sockpath = self . socket , transport = '<STR_LIT>' , timeout = self . _timeout ) <EOL> def _is_valid_executable ( self , binary_path ) : <EOL> return os . path . isfile ( binary_path ) and os . access ( binary_path , os . X_OK ) <EOL> def _normalize_watchman_path ( self , watchman_path ) : <EOL> if not self . _is_valid_executable ( watchman_path ) : <EOL> raise self . ExecutionError ( '<STR_LIT>' . format ( watchman_path ) ) <EOL> return os . path . abspath ( watchman_path ) <EOL> def _maybe_init_metadata ( self ) : <EOL> self . _logger . debug ( '<STR_LIT>' . format ( self . _watchman_work_dir ) ) <EOL> safe_mkdir ( self . _watchman_work_dir ) <EOL> safe_file_dump ( self . _state_file , '<STR_LIT:{}>' ) <EOL> def _construct_cmd ( self , cmd_parts , state_file , sock_file , log_file , log_level ) : <EOL> return [ part for part in cmd_parts ] + [ '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( state_file ) , <EOL> '<STR_LIT>' . format ( sock_file ) , <EOL> '<STR_LIT>' . format ( log_file ) , <EOL> '<STR_LIT>' , log_level ] <EOL> def _parse_pid_from_output ( self , output ) : <EOL> try : <EOL> return json . loads ( output ) [ '<STR_LIT>' ] <EOL> except ValueError : <EOL> self . _logger . critical ( '<STR_LIT>' . format ( output = output ) ) <EOL> raise self . InvalidCommandOutput ( output ) <EOL> except KeyError : <EOL> self . _logger . critical ( '<STR_LIT>' ) <EOL> raise self . InvalidCommandOutput ( output ) <EOL> def launch ( self ) : <EOL> """<STR_LIT>""" <EOL> cmd = self . _construct_cmd ( ( self . _watchman_path , '<STR_LIT>' ) , <EOL> state_file = self . _state_file , <EOL> sock_file = self . _sock_file , <EOL> log_file = self . _log_file , <EOL> log_level = str ( self . _log_level ) ) <EOL> self . _logger . debug ( '<STR_LIT>' . format ( '<STR_LIT:U+0020>' . join ( cmd ) ) ) <EOL> self . _maybe_init_metadata ( ) <EOL> get_output = functools . partial ( self . get_subprocess_output , cmd ) <EOL> output = retry_on_exception ( get_output , <NUM_LIT:3> , ( self . ExecutionError , ) , lambda n : n * <NUM_LIT> ) <EOL> pid = self . _parse_pid_from_output ( output ) <EOL> self . write_pid ( pid ) <EOL> self . write_socket ( self . _sock_file ) <EOL> def watch_project ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return self . client . query ( '<STR_LIT>' , os . path . realpath ( path ) ) <EOL> def subscribed ( self , build_root , handlers ) : <EOL> """<STR_LIT>""" <EOL> command_list = [ [ '<STR_LIT>' , <EOL> build_root , <EOL> handler . name , <EOL> handler . metadata ] for handler in handlers ] <EOL> self . _logger . debug ( '<STR_LIT>' . format ( command_list ) ) <EOL> for event in self . client . stream_query ( command_list ) : <EOL> if event is None : <EOL> yield None , None <EOL> elif '<STR_LIT>' in event : <EOL> self . _logger . info ( '<STR_LIT>' . format ( event ) ) <EOL> yield None , None <EOL> elif '<STR_LIT>' in event : <EOL> yield event . get ( '<STR_LIT>' ) , event <EOL> else : <EOL> self . _logger . warning ( '<STR_LIT>' . format ( event ) ) </s>
