text:
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap};
use std::convert::TryInto;
use std::io::Cursor;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::SystemTime;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use bazel_protos::gen::build::bazel::remote::execution::v2 as remexec;
use bazel_protos::gen::google::longrunning::Operation;
use bazel_protos::gen::google::rpc::{PreconditionFailure, Status as StatusProto};
use bazel_protos::require_digest;
use bytes::Bytes;
use concrete_time::TimeSpan;
use double_checked_cell_async::DoubleCheckedCell;
use fs::{self, File, PathStat};
use futures::future::{self, BoxFuture, TryFutureExt};
use futures::FutureExt;
use futures::{Stream, StreamExt};
use grpc_util::headers_to_http_header_map;
use grpc_util::prost::MessageExt;
use grpc_util::retry::{retry_call, status_is_retryable};
use grpc_util::{layered_service, status_to_str, LayeredService};
use hashing::{Digest, Fingerprint};
use log::{debug, trace, warn, Level};
use prost::Message;
use rand::{thread_rng, Rng};
use remexec::{
  action_cache_client::ActionCacheClient, capabilities_client::CapabilitiesClient,
  execution_client::ExecutionClient, Action, Command, ExecuteRequest, ExecuteResponse,
  ExecutedActionMetadata, ServerCapabilities, WaitExecutionRequest,
};
use store::{Snapshot, SnapshotOps, Store, StoreFileByDigest};
use tonic::metadata::BinaryMetadataValue;
use tonic::{Code, Request, Status};
use tryfuture::try_future;
use uuid::Uuid;
use workunit_store::{
  in_workunit, Metric, ObservationMetric, RunningWorkunit, SpanId, WorkunitMetadata, WorkunitStore,
};

use crate::{
  Context, FallibleProcessResultWithPlatform, MultiPlatformProcess, Platform, Process,
  ProcessCacheScope, ProcessMetadata, ProcessResultMetadata, ProcessResultSource,
};

// Environment variable which is exclusively used for cache key invalidation.
// This may be not specified in an Process, and may be populated only by the
// CommandRunner.
pub const CACHE_KEY_GEN_VERSION_ENV_VAR_NAME: &str = "PANTS_CACHE_KEY_GEN_VERSION";

// Environment variable which is used to include a unique value for cache busting of processes that
// have indicated that they should never be cached.
pub const CACHE_KEY_SALT_ENV_VAR_NAME: &str = "PANTS_CACHE_KEY_SALT";

// Environment variable which is exclusively used for cache key invalidation.
// This may be not specified in an Process, and may be populated only by the
// CommandRunner.
pub const CACHE_KEY_TARGET_PLATFORM_ENV_VAR_NAME: &str = "PANTS_CACHE_KEY_TARGET_PLATFORM";

#[derive(Debug)]
pub enum OperationOrStatus {
  Operation(Operation),
  Status(StatusProto),
}

#[derive(Debug, PartialEq)]
pub enum ExecutionError {
  // String is the error message.
  Fatal(String),
  // Digests are Files and Directories which have been reported to be missing. May be incomplete.
  MissingDigests(Vec<Digest>),
  // The server indicated that the request hit a timeout. Generally this is the timeout that the
  // client has pushed down on the ExecutionRequest.
  Timeout,
  // String is the error message.
  Retryable(String),
}

/// Implementation of CommandRunner that runs a command via the Bazel Remote Execution API
/// (https://docs.google.com/document/d/1AaGk7fOPByEvpAbqeXIyE8HX_A3_axxNnvroblTZ_6s/edit).
///
/// Results are streamed from the output stream of the Execute function (and possibly the
/// WaitExecution function if `CommandRunner` needs to reconnect).
///
/// If the CommandRunner has a Store, files will be uploaded to the remote CAS as needed.
/// Note that it does not proactively upload files to a remote CAS. This is because if we will
/// get a cache hit, uploading the files was wasted time and bandwidth, and if the remote CAS
/// already has some files, uploading them all is a waste. Instead, we look at the responses we
/// get back from the server, and upload the files it says it's missing.
///
/// In the future, we may want to do some clever things like proactively upload files which the
/// user has changed, or files which aren't known to the local git repository, but these are
/// optimizations to shave off a round-trip in the future.
#[derive(Clone)]
pub struct CommandRunner {
  metadata: ProcessMetadata,
  platform: Platform,
  store: Store,
  headers: BTreeMap<String, String>,
  execution_client: Arc<ExecutionClient<LayeredService>>,
  action_cache_client: Arc<ActionCacheClient<LayeredService>>,
  overall_deadline: Duration,
  retry_interval_duration: Duration,
  capabilities_cell: Arc<DoubleCheckedCell<ServerCapabilities>>,
  capabilities_client: Arc<CapabilitiesClient<LayeredService>>,
}

enum StreamOutcome {
  Complete(OperationOrStatus),
  StreamClosed(Option<String>),
}

impl CommandRunner {
  /// Construct a new CommandRunner
  pub fn new(
    execution_address: &str,
    store_address: &str,
    metadata: ProcessMetadata,
    root_ca_certs: Option<Vec<u8>>,
    headers: BTreeMap<String, String>,
    store: Store,
    platform: Platform,
    overall_deadline: Duration,
    retry_interval_duration: Duration,
    execution_concurrency_limit: usize,
    cache_concurrency_limit: usize,
    capabilities_cell_opt: Option<Arc<DoubleCheckedCell<ServerCapabilities>>>,
  ) -> Result<Self, String> {
    let execution_use_tls = execution_address.starts_with("https://");
    let store_use_tls = store_address.starts_with("https://");

    let tls_client_config = if execution_use_tls || store_use_tls {
      Some(grpc_util::tls::Config::new_without_mtls(root_ca_certs).try_into()?)
    } else {
      None
    };

    let mut execution_headers = headers.clone();
    let execution_endpoint = grpc_util::create_endpoint(
      execution_address,
      tls_client_config.as_ref().filter(|_| execution_use_tls),
      &mut execution_headers,
    )?;
    let execution_http_headers = headers_to_http_header_map(&execution_headers)?;
    let execution_channel = layered_service(
      tonic::transport::Channel::balance_list(vec![execution_endpoint].into_iter()),
      execution_concurrency_limit,
      execution_http_headers,
    );
    let execution_client = Arc::new(ExecutionClient::new(execution_channel.clone()));

    let mut store_headers = headers.clone();
    let store_endpoint = grpc_util::create_endpoint(
      store_address,
      tls_client_config.as_ref().filter(|_| execution_use_tls),
      &mut store_headers,
    )?;
    let store_http_headers = headers_to_http_header_map(&store_headers)?;
    let store_channel = layered_service(
      tonic::transport::Channel::balance_list(vec![store_endpoint].into_iter()),
      cache_concurrency_limit,
      store_http_headers,
    );

    let action_cache_client = Arc::new(ActionCacheClient::new(store_channel));

    let capabilities_client = Arc::new(CapabilitiesClient::new(execution_channel));

    let command_runner = CommandRunner {
      metadata,
      headers,
      execution_client,
      action_cache_client,
      store,
      platform,
      overall_deadline,
      retry_interval_duration,
      capabilities_cell: capabilities_cell_opt
        .unwrap_or_else(|| Arc::new(DoubleCheckedCell::new())),
      capabilities_client,
    };

    Ok(command_runner)
  }

  pub fn platform(&self) -> Platform {
    self.platform
  }

  async fn get_capabilities(&self) -> Result<&remexec::ServerCapabilities, String> {
    let capabilities_fut = async {
      let mut request = remexec::GetCapabilitiesRequest::default();
      if let Some(s) = self.metadata.instance_name.as_ref() {
        request.instance_name = s.clone();
      }

      let request = apply_headers(Request::new(request), "");

      let mut client = self.capabilities_client.as_ref().clone();
      client
        .get_capabilities(request)
        .await
        .map(|r| r.into_inner())
        .map_err(status_to_str)
    };

    self
      .capabilities_cell
      .get_or_try_init(capabilities_fut)
      .await
  }

  // Monitors the operation stream returned by the REv2 Execute and WaitExecution methods.
  // Outputs progress reported by the server and returns the next actionable operation
  // or gRPC status back to the main loop (plus the operation name so the main loop can
  // reconnect).
  async fn wait_on_operation_stream<S>(&self, mut stream: S, context: &Context) -> StreamOutcome
  where
    S: Stream<Item = Result<Operation, Status>> + Unpin,
  {
    let mut operation_name_opt: Option<String> = None;
    let mut start_time_opt = Some(Instant::now());

    trace!(
      "wait_on_operation_stream (build_id={}): monitoring stream",
      &context.build_id
    );

    loop {
      let item = stream.next().await;

      if let Some(start_time) = start_time_opt.take() {
        let timing: Result<u64, _> = Instant::now()
          .duration_since(start_time)
          .as_micros()
          .try_into();
        if let Ok(obs) = timing {
          context
            .workunit_store
            .record_observation(ObservationMetric::RemoteExecutionRPCFirstResponseTime, obs);
        }
      }

      match item {
        Some(Ok(operation)) => {
          trace!(
            "wait_on_operation_stream (build_id={}): got operation: {:?}",
            &context.build_id,
            &operation
          );

          // Extract the operation name.
          // Note: protobuf can return empty string for an empty field so convert empty strings
          // to None.
          operation_name_opt = Some(operation.name.clone()).filter(|s| !s.trim().is_empty());

          // Continue monitoring if the operation is not complete.
          if !operation.done {
            continue;
          }

          // Otherwise, return to the main loop with the operation as the result.
          return StreamOutcome::Complete(OperationOrStatus::Operation(operation));
        }

        Some(Err(err)) => {
          debug!("wait_on_operation_stream: got error: {:?}", err);
          let status_proto = StatusProto {
            code: err.code() as i32,
            message: err.message().to_string(),
            ..StatusProto::default()
          };
          return StreamOutcome::Complete(OperationOrStatus::Status(status_proto));
        }

        None => {
          // Stream disconnected unexpectedly.
          debug!("wait_on_operation_stream: unexpected disconnect from RE server");
          return StreamOutcome::StreamClosed(operation_name_opt);
        }
      }
    }
  }

  // Store the remote timings into the workunit store.
  fn save_workunit_timings(
    &self,
    execute_response: &ExecuteResponse,
    metadata: &ExecutedActionMetadata,
  ) {
    let workunit_thread_handle = workunit_store::expect_workunit_store_handle();
    let workunit_store = workunit_thread_handle.store;
    let parent_id = workunit_thread_handle.parent_id;
    let result_cached = execute_response.cached_result;

    if let (Some(queued_timestamp), Some(worker_start_timestamp)) = (
      metadata.queued_timestamp.as_ref(),
      metadata.worker_start_timestamp.as_ref(),
    ) {
      let span_result =
        TimeSpan::from_start_and_end(queued_timestamp, worker_start_timestamp, "remote queue");
      match span_result {
        Ok(time_span) => maybe_add_workunit(
          result_cached,
          "remote execution action scheduling",
          time_span,
          parent_id,
          &workunit_store,
          WorkunitMetadata {
            level: Level::Trace,
            ..WorkunitMetadata::default()
          },
        ),
        Err(s) => warn!("{}", s),
      }
    }

    if let (Some(input_fetch_start_timestamp), Some(input_fetch_completed_timestamp)) = (
      metadata.input_fetch_start_timestamp.as_ref(),
      metadata.input_fetch_completed_timestamp.as_ref(),
    ) {
      let span_result = TimeSpan::from_start_and_end(
        input_fetch_start_timestamp,
        input_fetch_completed_timestamp,
        "remote input fetch",
      );
      match span_result {
        Ok(time_span) => maybe_add_workunit(
          result_cached,
          "remote execution worker input fetching",
          time_span,
          parent_id,
          &workunit_store,
          WorkunitMetadata {
            level: Level::Trace,
            ..WorkunitMetadata::default()
          },
        ),
        Err(s) => warn!("{}", s),
      }
    }

    if let (Some(execution_start_timestamp), Some(execution_completed_timestamp)) = (
      metadata.execution_start_timestamp.as_ref(),
      metadata.execution_completed_timestamp.as_ref(),
    ) {
      let span_result = TimeSpan::from_start_and_end(
        execution_start_timestamp,
        execution_completed_timestamp,
        "remote execution",
      );
      match span_result {
        Ok(time_span) => maybe_add_workunit(
          result_cached,
          "remote execution worker command executing",
          time_span,
          parent_id,
          &workunit_store,
          WorkunitMetadata {
            level: Level::Trace,
            ..WorkunitMetadata::default()
          },
        ),
        Err(s) => warn!("{}", s),
      }
    }

    if let (Some(output_upload_start_timestamp), Some(output_upload_completed_timestamp)) = (
      metadata.output_upload_start_timestamp.as_ref(),
      metadata.output_upload_completed_timestamp.as_ref(),
    ) {
      let span_result = TimeSpan::from_start_and_end(
        output_upload_start_timestamp,
        output_upload_completed_timestamp,
        "remote output store",
      );
      match span_result {
        Ok(time_span) => maybe_add_workunit(
          result_cached,
          "remote execution worker output uploading",
          time_span,
          parent_id,
          &workunit_store,
          WorkunitMetadata {
            level: Level::Trace,
            ..WorkunitMetadata::default()
          },
        ),
        Err(s) => warn!("{}", s),
      }
    }
  }

  fn extract_missing_digests(&self, precondition_failure: &PreconditionFailure) -> ExecutionError {
    let mut missing_digests = Vec::with_capacity(precondition_failure.violations.len());

    for violation in &precondition_failure.violations {
      if violation.r#type != "MISSING" {
        return ExecutionError::Fatal(format!(
          "Unknown PreconditionFailure violation: {:?}",
          violation
        ));
      }

      let parts: Vec<_> = violation.subject.split('/').collect();
      if parts.len() != 3 || parts[0] != "blobs" {
        return ExecutionError::Fatal(format!(
          "Received FailedPrecondition MISSING but didn't recognize subject {}",
          violation.subject
        ));
      }

      let fingerprint = match Fingerprint::from_hex_string(parts[1]) {
        Ok(f) => f,
        Err(e) => {
          return ExecutionError::Fatal(format!("Bad digest in missing blob: {}: {}", parts[1], e))
        }
      };

      let size = match parts[2].parse::<usize>() {
        Ok(s) => s,
        Err(e) => {
          return ExecutionError::Fatal(format!("Missing blob had bad size: {}: {}", parts[2], e))
        }
      };

      missing_digests.push(Digest::new(fingerprint, size));
    }

    if missing_digests.is_empty() {
      return ExecutionError::Fatal(
        "Error from remote execution: FailedPrecondition, but no details".to_owned(),
      );
    }

    ExecutionError::MissingDigests(missing_digests)
  }

  // pub(crate) for testing
  pub(crate) async fn extract_execute_response(
    &self,
    operation_or_status: OperationOrStatus,
  ) -> Result<FallibleProcessResultWithPlatform, ExecutionError> {
    trace!("Got operation response: {:?}", operation_or_status);

    let status = match operation_or_status {
      OperationOrStatus::Operation(operation) => {
        assert!(operation.done, "operation was not marked done");

        use bazel_protos::gen::google::longrunning::operation::Result as OperationResult;
        let execute_response = match operation.result {
          Some(OperationResult::Response(response_any)) => {
            remexec::ExecuteResponse::decode(&response_any.value[..])
              .map_err(|e| ExecutionError::Fatal(format!("Invalid ExecuteResponse: {:?}", e)))?
          }
          Some(OperationResult::Error(rpc_status)) => {
            warn!("protocol violation: REv2 prohibits setting Operation::error");
            return Err(ExecutionError::Fatal(format_error(&rpc_status)));
          }
          None => {
            return Err(ExecutionError::Fatal(
              "Operation finished but no response supplied".to_string(),
            ));
          }
        };

        debug!("Got (nested) execute response: {:?}", execute_response);

        if let Some(ref metadata) = execute_response
          .result
          .as_ref()
          .and_then(|ar| ar.execution_metadata.clone())
        {
          self.save_workunit_timings(&execute_response, metadata);
        }

        let rpc_status = execute_response.status.unwrap_or_default();
        if rpc_status.code == Code::Ok as i32 {
          let action_result = if let Some(ref action_result) = execute_response.result {
            action_result
          } else {
            warn!("REv2 protocol violation: action result not set");
            return Err(ExecutionError::Fatal(
              "REv2 protocol violation: action result not set".into(),
            ));
          };

          return populate_fallible_execution_result(
            self.store.clone(),
            action_result,
            self.platform,
            false,
            if execute_response.cached_result {
              ProcessResultSource::HitRemotely
            } else {
              ProcessResultSource::RanRemotely
            },
          )
          .await
          .map_err(ExecutionError::Fatal);
        }

        rpc_status
      }
      OperationOrStatus::Status(status) => status,
    };

    match Code::from_i32(status.code) {
      Code::Ok => unreachable!(),

      Code::DeadlineExceeded => Err(ExecutionError::Timeout),

      Code::FailedPrecondition => {
        let details = if status.details.is_empty() {
          return Err(ExecutionError::Fatal(status.message));
        } else if status.details.len() > 1 {
          // TODO(tonic): Should we be able to handle multiple details protos?
          return Err(ExecutionError::Fatal(
            "too many detail protos for precondition failure".into(),
          ));
        } else {
          &status.details[0]
        };

        let full_name = format!("type.googleapis.com/{}", "google.rpc.PreconditionFailure");
        if details.type_url != full_name {
          return Err(ExecutionError::Fatal(format!(
            "Received PreconditionFailure, but didn't know how to resolve it: {}, protobuf type {}",
            status.message, details.type_url
          )));
        }

        // Decode the precondition failure.
        let precondition_failure = PreconditionFailure::decode(Cursor::new(&details.value))
          .map_err(|e| {
            ExecutionError::Fatal(format!(
              "Error deserializing PreconditionFailure proto: {:?}",
              e
            ))
          })?;

        Err(self.extract_missing_digests(&precondition_failure))
      }

      Code::Aborted
      | Code::Internal
      | Code::ResourceExhausted
      | Code::Unavailable
      | Code::Unknown => Err(ExecutionError::Retryable(status.message)),
      code => Err(ExecutionError::Fatal(format!(
        "Error from remote execution: {:?}: {:?}",
        code, status.message,
      ))),
    }
  }

  // Main loop: This function connects to the RE server and submits the given remote execution
  // request via the REv2 Execute method. It then monitors the operation stream until the
  // request completes. It will reconnect using the REv2 WaitExecution method if the connection
  // is dropped.
  //
  // The `run` method on CommandRunner uses this function to implement the bulk of the
  // processing for remote execution requests. The `run` method wraps the call with the method
  // with an overall deadline timeout.
  async fn run_execute_request(
    &self,
    execute_request: ExecuteRequest,
    process: Process,
    context: &Context,
    workunit: &mut RunningWorkunit,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    const MAX_RETRIES: u32 = 5;
    const MAX_BACKOFF_DURATION: Duration = Duration::from_secs(10);

    let start_time = Instant::now();
    let mut current_operation_name: Option<String> = None;
    let mut num_retries = 0;

    loop {
      // If we are currently retrying a request, then delay using an exponential backoff.
      if num_retries > 0 {
        workunit.increment_counter(Metric::RemoteExecutionRPCRetries, 1);

        let multiplier = thread_rng().gen_range(0..2_u32.pow(num_retries) + 1);
        let sleep_time = self.retry_interval_duration * multiplier;
        let sleep_time = sleep_time.min(MAX_BACKOFF_DURATION);
        debug!("delaying {:?} before retry", sleep_time);
        tokio::time::sleep(sleep_time).await;
      }

      let rpc_result = match current_operation_name {
        None => {
          // The request has not been submitted yet. Submit the request using the REv2
          // Execute method.
          debug!(
            "no current operation: submitting execute request: build_id={}; execute_request={:?}",
            context.build_id, &execute_request
          );
          workunit.increment_counter(Metric::RemoteExecutionRPCExecute, 1);
          let mut client = self.execution_client.as_ref().clone();
          let request = apply_headers(Request::new(execute_request.clone()), &context.build_id);
          client.execute(request).await
        }

        Some(ref operation_name) => {
          // The request has been submitted already. Reconnect to the status stream
          // using the REv2 WaitExecution method.
          debug!(
            "existing operation: reconnecting to operation stream: build_id={}; operation_name={}",
            context.build_id, operation_name
          );
          workunit.increment_counter(Metric::RemoteExecutionRPCWaitExecution, 1);
          let wait_execution_request = WaitExecutionRequest {
            name: operation_name.to_owned(),
          };
          let mut client = self.execution_client.as_ref().clone();
          let request = apply_headers(Request::new(wait_execution_request), &context.build_id);
          client.wait_execution(request).await
        }
      };

      // Take action based on whether we received an output stream or whether there is an
      // error to resolve.
      let actionable_result = match rpc_result {
        Ok(operation_stream_response) => {
          // Monitor the operation stream until there is an actionable operation
          // or status to interpret.
          let operation_stream = operation_stream_response.into_inner();
          let stream_outcome = self
            .wait_on_operation_stream(operation_stream, context)
            .await;

          match stream_outcome {
            StreamOutcome::Complete(status) => {
              trace!(
                "wait_on_operation_stream (build_id={}) returned completion={:?}",
                context.build_id,
                status
              );
              status
            }
            StreamOutcome::StreamClosed(operation_name_opt) => {
              trace!("wait_on_operation_stream (build_id={}) returned stream close, will retry operation_name={:?}", context.build_id, operation_name_opt);

              // Check if the number of request attempts sent thus far have exceeded the number
              // of retries allowed since the last successful connection. (There is no point in
              // continually submitting a request if ultimately futile.)
              if num_retries >= MAX_RETRIES {
                workunit.increment_counter(Metric::RemoteExecutionRPCErrors, 1);
                return Err(
                  "Too many failures from server. The last event was the server disconnecting with no error given.".to_owned(),
                );
              } else {
                // Increment the retry counter and allow loop to retry.
                num_retries += 1;
              }

              // Iterate the loop to reconnect to the operation.
              current_operation_name = operation_name_opt;
              continue;
            }
          }
        }
        Err(status) => {
          let status_proto = StatusProto {
            code: status.code() as i32,
            message: status.message().to_owned(),
            ..StatusProto::default()
          };
          OperationOrStatus::Status(status_proto)
        }
      };

      match self.extract_execute_response(actionable_result).await {
        Ok(result) => return Ok(result),
        Err(err) => match err {
          ExecutionError::Fatal(e) => {
            workunit.increment_counter(Metric::RemoteExecutionRPCErrors, 1);
            return Err(e);
          }
          ExecutionError::Retryable(e) => {
            // Check if the number of request attempts sent thus far have exceeded the number
            // of retries allowed since the last successful connection. (There is no point in
            // continually submitting a request if ultimately futile.)
            trace!("retryable error: {}", e);
            if num_retries >= MAX_RETRIES {
              workunit.increment_counter(Metric::RemoteExecutionRPCErrors, 1);
              return Err(format!(
                "Too many failures from server. The last error was: {}",
                e
              ));
            } else {
              // Increment the retry counter and allow loop to retry.
              num_retries += 1;
            }
          }
          ExecutionError::MissingDigests(missing_digests) => {
            trace!(
              "Server reported missing digests; trying to upload: {:?}",
              missing_digests,
            );

            let _ = self
              .store
              .ensure_remote_has_recursive(missing_digests)
              .await?;
          }
          ExecutionError::Timeout => {
            workunit.increment_counter(Metric::RemoteExecutionTimeouts, 1);
            return populate_fallible_execution_result_for_timeout(
              &self.store,
              &process.description,
              process.timeout,
              start_time.elapsed(),
              self.platform,
            )
            .await;
          }
        },
      }
    }
  }
}

#[async_trait]
impl crate::CommandRunner for CommandRunner {
  /// Run the given MultiPlatformProcess via the Remote Execution API.
  async fn run(
    &self,
    context: Context,
    _workunit: &mut RunningWorkunit,
    request: MultiPlatformProcess,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    // Retrieve capabilities for this server.
    let capabilities = self.get_capabilities().await?;
    trace!("RE capabilities: {:?}", &capabilities);

    // Construct the REv2 ExecuteRequest and related data for this execution request.
    let request = self.extract_compatible_request(&request).unwrap();
    let (action, command, execute_request) = make_execute_request(&request, self.metadata.clone())?;
    let build_id = context.build_id.clone();

    debug!("Remote execution: {}", request.description);
    debug!(
      "built REv2 request (build_id={}): action={:?}; command={:?}; execute_request={:?}",
      &build_id, action, command, execute_request
    );

    // Record the time that we started to process this request, then compute the ultimate
    // deadline for execution of this request.
    let deadline_duration = self.overall_deadline + request.timeout.unwrap_or_default();

    // Ensure the action and command are stored locally.
    let (command_digest, action_digest) =
      ensure_action_stored_locally(&self.store, &command, &action).await?;

    // Check the remote Action Cache to see if this request was already computed.
    // If so, return immediately with the result.
    let context2 = context.clone();
    let cached_response_opt = check_action_cache(
      action_digest,
      &request.description,
      &self.metadata,
      self.platform,
      &context2,
      self.action_cache_client.clone(),
      self.store.clone(),
      false,
    )
    .await?;
    debug!(
      "action cache response: build_id={}; digest={:?}: {:?}",
      &build_id, action_digest, cached_response_opt
    );
    if let Some(cached_response) = cached_response_opt {
      return Ok(cached_response);
    }

    // Upload the action (and related data, i.e. the embedded command and input files).
    ensure_action_uploaded(
      &context,
      &self.store,
      command_digest,
      action_digest,
      Some(request.input_files),
    )
    .await?;

    // Submit the execution request to the RE server for execution.
    let context2 = context.clone();
    in_workunit!(
      context.workunit_store.clone(),
      "run_execute_request".to_owned(),
      WorkunitMetadata {
        // NB: See engine::nodes::NodeKey::workunit_level for more information on why this workunit
        // renders at the Process's level.
        level: request.level,
        desc: Some(request.description.clone()),
        ..WorkunitMetadata::default()
      },
      |workunit| async move {
        workunit.increment_counter(Metric::RemoteExecutionRequests, 1);
        let result_fut = self.run_execute_request(execute_request, request, &context2, workunit);
        let result = tokio::time::timeout(deadline_duration, result_fut).await;
        if result.is_err() {
          workunit.update_metadata(|inititial| WorkunitMetadata {
            level: Level::Error,
            desc: Some(format!(
              "remote execution timed out after {:?}",
              deadline_duration
            )),
            ..inititial
          })
        }

        // Detect whether the operation ran or hit the deadline timeout.
        match result {
          Ok(result) => {
            if result.is_ok() {
              workunit.increment_counter(Metric::RemoteExecutionSuccess, 1);
            } else {
              workunit.increment_counter(Metric::RemoteExecutionErrors, 1);
            }
            result
          }
          Err(_) => {
            // The Err in this match arm originates from the timeout future.
            debug!(
              "remote execution for build_id={} timed out after {:?}",
              &build_id, deadline_duration
            );
            workunit.increment_counter(Metric::RemoteExecutionTimeouts, 1);
            Err(format!(
              "remote execution timed out after {:?}",
              deadline_duration
            ))
          }
        }
      },
    )
    .await
  }

  // TODO: This is a copy of the same method on crate::remote::CommandRunner.
  fn extract_compatible_request(&self, req: &MultiPlatformProcess) -> Option<Process> {
    for compatible_constraint in vec![None, self.platform.into()].iter() {
      if let Some(compatible_req) = req.0.get(compatible_constraint) {
        return Some(compatible_req.clone());
      }
    }
    None
  }
}

fn maybe_add_workunit(
  result_cached: bool,
  name: &str,
  time_span: concrete_time::TimeSpan,
  parent_id: Option<SpanId>,
  workunit_store: &WorkunitStore,
  metadata: WorkunitMetadata,
) {
  if !result_cached {
    let start_time: SystemTime = SystemTime::UNIX_EPOCH + time_span.start.into();
    let end_time: SystemTime = start_time + time_span.duration.into();
    workunit_store.add_completed_workunit(
      name.to_string(),
      start_time,
      end_time,
      parent_id,
      metadata,
    );
  }
}

pub fn make_execute_request(
  req: &Process,
  metadata: ProcessMetadata,
) -> Result<(remexec::Action, remexec::Command, remexec::ExecuteRequest), String> {
  let mut command = remexec::Command {
    arguments: req.argv.clone(),
    ..remexec::Command::default()
  };
  for (name, value) in &req.env {
    if name == CACHE_KEY_GEN_VERSION_ENV_VAR_NAME
      || name == CACHE_KEY_TARGET_PLATFORM_ENV_VAR_NAME
      || name == CACHE_KEY_SALT_ENV_VAR_NAME
    {
      return Err(format!(
        "Cannot set env var with name {} as that is reserved for internal use by pants",
        name
      ));
    }

    command
      .environment_variables
      .push(remexec::command::EnvironmentVariable {
        name: name.to_string(),
        value: value.to_string(),
      });
  }

  let ProcessMetadata {
    instance_name,
    cache_key_gen_version,
    mut platform_properties,
  } = metadata;

  // TODO: Disabling append-only caches in remoting until server support exists due to
  //       interaction with how servers match platform properties.
  // if !req.append_only_caches.is_empty() {
  //   platform_properties.extend(NamedCaches::platform_properties(
  //     &req.append_only_caches,
  //     &cache_key_gen_version,
  //   ));
  // }

  if let Some(cache_key_gen_version) = cache_key_gen_version {
    command
      .environment_variables
      .push(remexec::command::EnvironmentVariable {
        name: CACHE_KEY_GEN_VERSION_ENV_VAR_NAME.to_string(),
        value: cache_key_gen_version,
      });
  }

  if matches!(
    req.cache_scope,
    ProcessCacheScope::PerSession
      | ProcessCacheScope::PerRestartAlways
      | ProcessCacheScope::PerRestartSuccessful
  ) {
    command
      .environment_variables
      .push(remexec::command::EnvironmentVariable {
        name: CACHE_KEY_SALT_ENV_VAR_NAME.to_string(),
        value: Uuid::new_v4().to_string(),
      });
  }

  {
    command
      .environment_variables
      .push(remexec::command::EnvironmentVariable {
        name: CACHE_KEY_TARGET_PLATFORM_ENV_VAR_NAME.to_string(),
        value: match req.platform_constraint {
          Some(plat) => plat.into(),
          None => "none".to_string(),
        },
      });
  }

  let mut output_files = req
    .output_files
    .iter()
    .map(|p| {
      p.to_str()
        .map(str::to_owned)
        .ok_or_else(|| format!("Non-UTF8 output file path: {:?}", p))
    })
    .collect::<Result<Vec<String>, String>>()?;
  output_files.sort();
  command.output_files = output_files;

  let mut output_directories = req
    .output_directories
    .iter()
    .map(|p| {
      p.to_str()
        .map(str::to_owned)
        .ok_or_else(|| format!("Non-UTF8 output directory path: {:?}", p))
    })
    .collect::<Result<Vec<String>, String>>()?;
  output_directories.sort();
  command.output_directories = output_directories;

  if let Some(working_directory) = &req.working_directory {
    command.working_directory = working_directory
      .to_str()
      .map(str::to_owned)
      .unwrap_or_else(|| panic!("Non-UTF8 working directory path: {:?}", working_directory));
  }

  if req.jdk_home.is_some() {
    // Ideally, the JDK would be brought along as part of the input directory, but we don't
    // currently have support for that. Scoot supports this property, and will symlink .jdk to a
    // system-installed JDK https://github.com/twitter/scoot/pull/391 - we should probably come to
    // some kind of consensus across tools as to how this should work; RBE appears to work by
    // allowing you to specify a jdk-version platform property, and it will put a JDK at a
    // well-known path in the docker container you specify in which to run.
    platform_properties.push(("JDK_SYMLINK".to_owned(), ".jdk".to_owned()));
  }

  // Extract `Platform` proto from the `Command` to avoid a partial move of `Command`.
  let mut command_platform = command.platform.take().unwrap_or_default();

  // Add configured platform properties to the `Platform`.
  for (name, value) in platform_properties {
    command_platform
      .properties
      .push(remexec::platform::Property {
        name: name.clone(),
        value: value.clone(),
      });
  }

  // Sort the platform properties.
  //
  // From the remote execution spec:
  //   The properties that make up this platform. In order to ensure that
  //   equivalent `Platform`s always hash to the same value, the properties MUST
  //   be lexicographically sorted by name, and then by value. Sorting of strings
  //   is done by code point, equivalently, by the UTF-8 bytes.
  //
  // Note: BuildBarn enforces this requirement.
  command_platform
    .properties
    .sort_by(|x, y| match x.name.cmp(&y.name) {
      Ordering::Equal => x.value.cmp(&y.value),
      v => v,
    });

  // Store the separate copy back into the Command proto.
  command.platform = Some(command_platform);

  // Sort the environment variables. REv2 spec requires sorting by name for same reasons that
  // platform properties are sorted, i.e. consistent hashing.
  command
    .environment_variables
    .sort_by(|x, y| x.name.cmp(&y.name));

  let mut action = remexec::Action {
    command_digest: Some((&digest(&command)?).into()),
    input_root_digest: Some((&req.input_files).into()),
    ..remexec::Action::default()
  };

  if let Some(timeout) = req.timeout {
    action.timeout = Some(prost_types::Duration::from(timeout));
  }

  let execute_request = remexec::ExecuteRequest {
    action_digest: Some((&digest(&action)?).into()),
    instance_name: instance_name.unwrap_or_else(|| "".to_owned()),
    ..remexec::ExecuteRequest::default()
  };

  Ok((action, command, execute_request))
}

pub async fn populate_fallible_execution_result_for_timeout(
  store: &Store,
  description: &str,
  timeout: Option<Duration>,
  elapsed: Duration,
  platform: Platform,
) -> Result<FallibleProcessResultWithPlatform, String> {
  let timeout_msg = if let Some(timeout) = timeout {
    format!("user timeout of {:?} after {:?}", timeout, elapsed)
  } else {
    format!("server timeout after {:?}", elapsed)
  };
  let stdout = Bytes::from(format!("Exceeded {} for {}", timeout_msg, description));
  let stdout_digest = store.store_file_bytes(stdout, true).await?;

  Ok(FallibleProcessResultWithPlatform {
    stdout_digest,
    stderr_digest: hashing::EMPTY_DIGEST,
    exit_code: -libc::SIGTERM,
    output_directory: hashing::EMPTY_DIGEST,
    platform,
    metadata: ProcessResultMetadata::new(Some(elapsed.into()), ProcessResultSource::RanRemotely),
  })
}

/// Convert an ActionResult into a FallibleProcessResultWithPlatform.
///
/// HACK: The caching CommandRunner stores the digest of the Directory that merges all output
/// files and output directories in the `tree_digest` field of the `output_directories` field
/// of the ActionResult/ExecuteResponse stored in the local cache. When
/// `treat_tree_digest_as_final_directory_hack` is true, then that final merged directory
/// will be extracted from the tree_digest of the single output directory.
pub fn populate_fallible_execution_result(
  store: Store,
  action_result: &remexec::ActionResult,
  platform: Platform,
  treat_tree_digest_as_final_directory_hack: bool,
  source: ProcessResultSource,
) -> BoxFuture<Result<FallibleProcessResultWithPlatform, String>> {
  future::try_join3(
    extract_stdout(&store, action_result),
    extract_stderr(&store, action_result),
    extract_output_files(
      store,
      action_result,
      treat_tree_digest_as_final_directory_hack,
    ),
  )
  .and_then(
    move |(stdout_digest, stderr_digest, output_directory)| async move {
      Ok(FallibleProcessResultWithPlatform {
        stdout_digest,
        stderr_digest,
        exit_code: action_result.exit_code,
        output_directory,
        platform,
        metadata: action_result
          .execution_metadata
          .clone()
          .map_or(ProcessResultMetadata::new(None, source), |metadata| {
            ProcessResultMetadata::new_from_metadata(metadata, source)
          }),
      })
    },
  )
  .boxed()
}

fn extract_stdout<'a>(
  store: &Store,
  action_result: &'a remexec::ActionResult,
) -> BoxFuture<'a, Result<Digest, String>> {
  let store = store.clone();
  async move {
    if let Some(digest_proto) = &action_result.stdout_digest {
      let stdout_digest_result: Result<Digest, String> = digest_proto.try_into();
      let stdout_digest =
        stdout_digest_result.map_err(|err| format!("Error extracting stdout: {}", err))?;
      Ok(stdout_digest)
    } else {
      let stdout_raw = Bytes::copy_from_slice(&action_result.stdout_raw);
      let digest = store
        .store_file_bytes(stdout_raw, true)
        .map_err(move |error| format!("Error storing raw stdout: {:?}", error))
        .await?;
      Ok(digest)
    }
  }
  .boxed()
}

fn extract_stderr<'a>(
  store: &Store,
  action_result: &'a remexec::ActionResult,
) -> BoxFuture<'a, Result<Digest, String>> {
  let store = store.clone();
  async move {
    if let Some(digest_proto) = &action_result.stderr_digest {
      let stderr_digest_result: Result<Digest, String> = digest_proto.try_into();
      let stderr_digest =
        stderr_digest_result.map_err(|err| format!("Error extracting stderr: {}", err))?;
      Ok(stderr_digest)
    } else {
      let stderr_raw = Bytes::copy_from_slice(&action_result.stderr_raw);
      let digest = store
        .store_file_bytes(stderr_raw, true)
        .map_err(move |error| format!("Error storing raw stderr: {:?}", error))
        .await?;
      Ok(digest)
    }
  }
  .boxed()
}

pub fn extract_output_files(
  store: Store,
  action_result: &remexec::ActionResult,
  treat_tree_digest_as_final_directory_hack: bool,
) -> BoxFuture<'static, Result<Digest, String>> {
  // HACK: The caching CommandRunner stores the digest of the Directory that merges all output
  // files and output directories in the `tree_digest` field of the `output_directories` field
  // of the ActionResult/ExecuteResponse stored in the local cache. When
  // `treat_tree_digest_as_final_directory_hack` is true, then this code will extract that
  // directory from the tree_digest and skip the merging performed by the remainder of this
  // method.
  if treat_tree_digest_as_final_directory_hack {
    match &action_result.output_directories[..] {
      &[ref directory] => {
        match require_digest(directory.tree_digest.as_ref()) {
          Ok(digest) => return future::ready::<Result<Digest, String>>(Ok(digest)).boxed(),
          Err(err) => return futures::future::err(err).boxed(),
        };
      }
      _ => {
        return futures::future::err(
          "illegal state: treat_tree_digest_as_final_directory_hack expected single output directory".to_owned()
        ).boxed();
      }
    }
  }

  // Get Digests of output Directories.
  // Then we'll make a Directory for the output files, and merge them.
  let mut directory_digests = Vec::with_capacity(action_result.output_directories.len() + 1);
  // TODO: Maybe take rather than clone
  let output_directories = action_result.output_directories.clone();
  for dir in output_directories {
    let store = store.clone();
    directory_digests.push(
      (async move {
        // The `OutputDirectory` contains the digest of a `Tree` proto which contains
        // the `Directory` proto of the root directory of this `OutputDirectory` plus all
        // of the `Directory` protos for child directories of that root.

        // Retrieve the Tree proto and hash its root `Directory` proto to obtain the digest
        // of the output directory needed to construct the series of `Directory` protos needed
        // for the final merge of the output directories.
        let tree_digest: Digest = require_digest(dir.tree_digest.as_ref())?;
        let root_digest_opt = store.load_tree_from_remote(tree_digest).await?;
        let root_digest = root_digest_opt
          .ok_or_else(|| format!("Tree with digest {:?} was not in remote", tree_digest))?;

        let mut digest = root_digest;

        if !dir.path.is_empty() {
          for component in dir.path.rsplit('/') {
            let component = component.to_owned();
            let directory = remexec::Directory {
              directories: vec![remexec::DirectoryNode {
                name: component,
                digest: Some((&digest).into()),
              }],
              ..remexec::Directory::default()
            };
            digest = store.record_directory(&directory, true).await?;
          }
        }
        let res: Result<_, String> = Ok(digest);
        res
      })
      .map_err(|err| format!("Error saving remote output directory: {}", err)),
    );
  }

  // Make a directory for the files
  let mut path_map = HashMap::new();
  let path_stats_result: Result<Vec<PathStat>, String> = action_result
    .output_files
    .iter()
    .map(|output_file| {
      let output_file_path_buf = PathBuf::from(output_file.path.clone());
      let digest: Result<Digest, String> = require_digest(output_file.digest.as_ref());
      path_map.insert(output_file_path_buf.clone(), digest?);
      Ok(PathStat::file(
        output_file_path_buf.clone(),
        File {
          path: output_file_path_buf,
          is_executable: output_file.is_executable,
        },
      ))
    })
    .collect();

  let path_stats = try_future!(path_stats_result);

  #[derive(Clone)]
  struct StoreOneOffRemoteDigest {
    map_of_paths_to_digests: HashMap<PathBuf, Digest>,
  }

  impl StoreOneOffRemoteDigest {
    fn new(map: HashMap<PathBuf, Digest>) -> StoreOneOffRemoteDigest {
      StoreOneOffRemoteDigest {
        map_of_paths_to_digests: map,
      }
    }
  }

  impl StoreFileByDigest<String> for StoreOneOffRemoteDigest {
    fn store_by_digest(&self, file: File) -> future::BoxFuture<'static, Result<Digest, String>> {
      match self.map_of_paths_to_digests.get(&file.path) {
        Some(digest) => future::ok(*digest),
        None => future::err(format!(
          "Didn't know digest for path in remote execution response: {:?}",
          file.path
        )),
      }
      .boxed()
    }
  }

  async move {
    let files_digest = Snapshot::digest_from_path_stats(
      store.clone(),
      StoreOneOffRemoteDigest::new(path_map),
      path_stats,
    )
    .map_err(move |error| {
      format!(
        "Error when storing the output file directory info in the remote CAS: {:?}",
        error
      )
    });

    let (files_digest, mut directory_digests) =
      future::try_join(files_digest, future::try_join_all(directory_digests)).await?;

    directory_digests.push(files_digest);

    store
      .merge(directory_digests)
      .map_err(|err| format!("Error when merging output files and directories: {:?}", err))
      .await
  }
  .boxed()
}

/// Apply REAPI request metadata header to a `tonic::Request`.
fn apply_headers<T>(mut request: Request<T>, build_id: &str) -> Request<T> {
  let reapi_request_metadata = remexec::RequestMetadata {
    tool_details: Some(remexec::ToolDetails {
      tool_name: "pants".into(),
      ..remexec::ToolDetails::default()
    }),
    tool_invocation_id: build_id.to_string(),
    ..remexec::RequestMetadata::default()
  };

  let md = request.metadata_mut();
  md.insert_bin(
    "google.devtools.remoteexecution.v1test.requestmetadata-bin",
    BinaryMetadataValue::try_from_bytes(&reapi_request_metadata.to_bytes()).unwrap(),
  );

  request
}

/// Check the remote Action Cache for a cached result of running the given `command` and the Action
/// with the given `action_digest`.
///
/// This check is necessary because some REAPI servers do not short-circuit the Execute method
/// by checking the Action Cache (e.g., BuildBarn). Thus, this client must check the cache
/// explicitly in order to avoid duplicating already-cached work. This behavior matches
/// the Bazel RE client.
pub async fn check_action_cache(
  action_digest: Digest,
  command_description: &str,
  metadata: &ProcessMetadata,
  platform: Platform,
  context: &Context,
  action_cache_client: Arc<ActionCacheClient<LayeredService>>,
  store: Store,
  eager_fetch: bool,
) -> Result<Option<FallibleProcessResultWithPlatform>, String> {
  in_workunit!(
    context.workunit_store.clone(),
    "check_action_cache".to_owned(),
    WorkunitMetadata {
      level: Level::Debug,
      desc: Some(format!("Remote cache lookup for: {}", command_description)),
      ..WorkunitMetadata::default()
    },
    |workunit| async move {
      workunit.increment_counter(Metric::RemoteCacheRequests, 1);

      let client = action_cache_client.as_ref().clone();
      let action_result_response = retry_call(
        client,
        move |mut client| {
          let request = remexec::GetActionResultRequest {
            action_digest: Some(action_digest.into()),
            instance_name: metadata
              .instance_name
              .as_ref()
              .cloned()
              .unwrap_or_else(String::new),
            ..remexec::GetActionResultRequest::default()
          };
          let request = apply_headers(Request::new(request), &context.build_id);
          async move { client.get_action_result(request).await }
        },
        status_is_retryable,
      )
      .await;

      match action_result_response {
        Ok(action_result) => {
          let action_result = action_result.into_inner();
          let response = populate_fallible_execution_result(
            store.clone(),
            &action_result,
            platform,
            false,
            ProcessResultSource::HitRemotely,
          )
          .await?;
          // TODO: This should move inside the retry_call above, both in order to be retried, and
          // to ensure that we increment a miss if we fail to eagerly fetch.
          if eager_fetch {
            let response = response.clone();
            in_workunit!(
              context.workunit_store.clone(),
              "eager_fetch_action_cache".to_owned(),
              WorkunitMetadata {
                level: Level::Trace,
                desc: Some("eagerly fetching after action cache hit".to_owned()),
                ..WorkunitMetadata::default()
              },
              |_workunit| async move {
                future::try_join_all(vec![
                  store.ensure_local_has_file(response.stdout_digest).boxed(),
                  store.ensure_local_has_file(response.stderr_digest).boxed(),
                  store
                    .ensure_local_has_recursive_directory(response.output_directory)
                    .boxed(),
                ])
                .await
              }
            )
            .await?;
          }
          workunit.increment_counter(Metric::RemoteCacheRequestsCached, 1);
          Ok(Some(response))
        }
        Err(status) => match status.code() {
          Code::NotFound => {
            workunit.increment_counter(Metric::RemoteCacheRequestsUncached, 1);
            Ok(None)
          }
          _ => {
            workunit.increment_counter(Metric::RemoteCacheReadErrors, 1);
            Err(status_to_str(status))
          }
        },
      }
    }
  )
  .await
}

pub async fn store_proto_locally<P: prost::Message>(
  store: &Store,
  proto: &P,
) -> Result<Digest, String> {
  store
    .store_file_bytes(proto.to_bytes(), true)
    .await
    .map_err(|e| format!("Error saving proto to local store: {:?}", e))
}

pub async fn ensure_action_stored_locally(
  store: &Store,
  command: &Command,
  action: &Action,
) -> Result<(Digest, Digest), String> {
  let (command_digest, action_digest) = future::try_join(
    store_proto_locally(store, command),
    store_proto_locally(store, action),
  )
  .await?;

  Ok((command_digest, action_digest))
}

///
/// Ensure that the Action and Command (and optionally their input files, likely depending on
/// whether we are in a remote execution context, or a pure cache-usage context) are uploaded.
///
pub async fn ensure_action_uploaded(
  context: &Context,
  store: &Store,
  command_digest: Digest,
  action_digest: Digest,
  input_files: Option<Digest>,
) -> Result<(), String> {
  in_workunit!(
    context.workunit_store.clone(),
    "ensure_action_uploaded".to_owned(),
    WorkunitMetadata {
      level: Level::Trace,
      desc: Some(format!("ensure action uploaded for {:?}", action_digest)),
      ..WorkunitMetadata::default()
    },
    |_workunit| async move {
      let mut digests = vec![command_digest, action_digest];
      digests.extend(input_files);
      let _ = store.ensure_remote_has_recursive(digests).await?;
      Ok(())
    },
  )
  .await
}

pub fn format_error(error: &StatusProto) -> String {
  let error_code_enum = Code::from_i32(error.code);
  let error_code = match error_code_enum {
    Code::Unknown => format!("{:?}", error.code),
    x => format!("{:?}", x),
  };
  format!("{}: {}", error_code, error.message)
}

pub fn digest<T: prost::Message>(message: &T) -> Result<Digest, String> {
  Ok(Digest::of_bytes(&message.to_bytes()))
}


text:
package org.apache.bcel.generic;

/**
 * Denotes that an instruction may start the process of loading and resolving the referenced class in the Virtual Machine.
 */
public interface LoadClass {

    /**
     * Returns the {@link ObjectType} of the referenced class or interface that may be loaded and resolved.
     *
     * @param cpg A ConstantPoolGen
     * @return object type that may be loaded or null if a primitive is referenced
     */
    ObjectType getLoadClassType(ConstantPoolGen cpg);

    /**
     * Returns the type associated with this instruction. LoadClass instances are always typed, but this type does not always refer to the type of the class or
     * interface that it possibly forces to load. For example, {@link GETFIELD} would return the type of the field and not the type of the class where the field
     * is defined. If no class is forced to be loaded, {@code null} is returned. An example for this is an {@link NEWARRAY} instruction that creates an
     * {@code int[][]}.
     *
     * @param cpg A ConstantPoolGen
     * @return the type associated with this instruction.
     * @see #getLoadClassType(ConstantPoolGen)
     */
    Type getType(ConstantPoolGen cpg);
}

redpajama_set_name: RedPajamaGithub


text:
package com.thoughtworks.go.config.serialization;

import com.thoughtworks.go.config.*;
import com.thoughtworks.go.config.materials.AbstractMaterialConfig;
import com.thoughtworks.go.config.materials.MaterialConfigs;
import com.thoughtworks.go.config.materials.dependency.DependencyMaterialConfig;
import com.thoughtworks.go.config.materials.perforce.P4MaterialConfig;
import com.thoughtworks.go.config.remote.FileConfigOrigin;
import com.thoughtworks.go.domain.ConfigErrors;
import com.thoughtworks.go.domain.materials.dependency.NewGoConfigMother;
import com.thoughtworks.go.helper.GoConfigMother;
import com.thoughtworks.go.util.ConfigElementImplementationRegistryMother;
import org.apache.commons.io.IOUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.HashMap;
import java.util.Map;

import static com.thoughtworks.go.helper.MaterialConfigsMother.p4;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.assertThat;

public class DependencyMaterialConfigTest {

    private MagicalGoConfigXmlWriter writer;
    private MagicalGoConfigXmlLoader loader;
    private CruiseConfig config;
    private PipelineConfig pipelineConfig;

    @Before
    public void setUp() throws Exception {
        writer = new MagicalGoConfigXmlWriter(new ConfigCache(), ConfigElementImplementationRegistryMother.withNoPlugins());
        loader = new MagicalGoConfigXmlLoader(new ConfigCache(), ConfigElementImplementationRegistryMother.withNoPlugins());
        config = GoConfigMother.configWithPipelines("pipeline1", "pipeline2", "pipeline3", "go");
        pipelineConfig = config.getAllPipelineConfigs().get(0);
    }

    @Test
    public void shouldBeAbleToLoadADependencyMaterialFromConfig() throws Exception {
        String xml = "<pipeline pipelineName=\"pipeline-name\" stageName=\"stage-name\" />";
        DependencyMaterialConfig material = loader.fromXmlPartial(xml, DependencyMaterialConfig.class);
        assertThat(material.getPipelineName(), is(new CaseInsensitiveString("pipeline-name")));
        assertThat(material.getStageName(), is(new CaseInsensitiveString("stage-name")));
        assertThat(writer.toXmlPartial(material), is(xml));
    }

    @Test
    public void shouldBeAbleToSaveADependencyMaterialToConfig() throws Exception {
        DependencyMaterialConfig originalMaterial = new DependencyMaterialConfig(new CaseInsensitiveString("pipeline-name"), new CaseInsensitiveString("stage-name"));

        NewGoConfigMother mother = new NewGoConfigMother();
        mother.addPipeline("pipeline-name", "stage-name", "job-name");
        mother.addPipeline("dependent", "stage-name", "job-name").addMaterialConfig(originalMaterial);
        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        CruiseConfig configForEdit = mother.cruiseConfig();
        configForEdit.initializeServer();
        writer.write(configForEdit, buffer, false);

        final ByteArrayInputStream inputStream = new ByteArrayInputStream(buffer.toByteArray());
        CruiseConfig config = loader.loadConfigHolder(IOUtils.toString(inputStream, UTF_8)).config;

        DependencyMaterialConfig material = (DependencyMaterialConfig) config.pipelineConfigByName(new CaseInsensitiveString("dependent")).materialConfigs().get(1);
        assertThat(material, is(originalMaterial));
        assertThat(material.getPipelineName(), is(new CaseInsensitiveString("pipeline-name")));
        assertThat(material.getStageName(), is(new CaseInsensitiveString("stage-name")));
    }

    @Test
    public void shouldBeAbleToHaveADependencyAndOneOtherMaterial() throws Exception {
        NewGoConfigMother mother = new NewGoConfigMother();
        mother.addPipeline("pipeline-name", "stage-name", "job-name");
        PipelineConfig pipelineConfig = mother.addPipeline("dependent", "stage-name", "job-name",
                new DependencyMaterialConfig(new CaseInsensitiveString("pipeline-name"), new CaseInsensitiveString("stage-name")));
        pipelineConfig.addMaterialConfig(p4("localhost:1666", "foo"));

        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        CruiseConfig cruiseConfig = mother.cruiseConfig();
        cruiseConfig.initializeServer();

        writer.write(cruiseConfig, buffer, false);

        final ByteArrayInputStream inputStream = new ByteArrayInputStream(buffer.toByteArray());
        CruiseConfig config = loader.loadConfigHolder(IOUtils.toString(inputStream, UTF_8)).config;

        MaterialConfigs materialConfigs = config.pipelineConfigByName(new CaseInsensitiveString("dependent")).materialConfigs();
        assertThat(materialConfigs.get(0), is(instanceOf(DependencyMaterialConfig.class)));
        assertThat(materialConfigs.get(1), is(instanceOf(P4MaterialConfig.class)));
    }

    @Test
    public void shouldAddErrorForInvalidMaterialName() {
        DependencyMaterialConfig materialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("wrong name"), new CaseInsensitiveString("pipeline-foo"), new CaseInsensitiveString("stage-bar"));
        materialConfig.validate(ConfigSaveValidationContext.forChain(new BasicCruiseConfig(), pipelineConfig));
        assertThat(materialConfig.errors().on(AbstractMaterialConfig.MATERIAL_NAME), is("Invalid material name 'wrong name'. This must be alphanumeric and can contain underscores and periods (however, it cannot start with a period). The maximum allowed length is 255 characters."));
    }

    @Test
    public void shouldAddErrorWhenInvalidPipelineNameStage() {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig();
        Map<String, String> configMap = new HashMap<>();
        configMap.put(DependencyMaterialConfig.PIPELINE_STAGE_NAME, "invalid pipeline stage");

        dependencyMaterialConfig.setConfigAttributes(configMap);

        assertThat(dependencyMaterialConfig.getPipelineStageName(), is("invalid pipeline stage"));
        assertThat(dependencyMaterialConfig.errors().isEmpty(), is(false));
        assertThat(dependencyMaterialConfig.errors().on(DependencyMaterialConfig.PIPELINE_STAGE_NAME), is("'invalid pipeline stage' should conform to the pattern 'pipeline [stage]'"));
    }

    @Test
    public void shouldNotBombValidationWhenMaterialNameIsNotSet() {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("pipeline-foo"), new CaseInsensitiveString("stage-bar"));
        dependencyMaterialConfig.validate(ConfigSaveValidationContext.forChain(new BasicCruiseConfig(), pipelineConfig));
        assertThat(dependencyMaterialConfig.errors().on(AbstractMaterialConfig.MATERIAL_NAME), is(nullValue()));
    }

    @Test
    public void shouldNOTBeValidIfThePipelineExistsButTheStageDoesNot() throws Exception {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("pipeline2"), new CaseInsensitiveString("stage-not-existing does not exist!"));
        dependencyMaterialConfig.validate(ConfigSaveValidationContext.forChain(config, pipelineConfig));
        ConfigErrors configErrors = dependencyMaterialConfig.errors();
        assertThat(configErrors.isEmpty(), is(false));
        assertThat(configErrors.on(DependencyMaterialConfig.PIPELINE_STAGE_NAME), containsString("Stage with name 'stage-not-existing does not exist!' does not exist on pipeline 'pipeline2'"));
    }

    @Test
    public void shouldNOTBeValidIfTheReferencedPipelineDoesNotExist() throws Exception {
        CruiseConfig config = GoConfigMother.configWithPipelines("pipeline1", "pipeline2", "pipeline3", "go");

        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("pipeline-not-exist"), new CaseInsensitiveString("stage"));
        dependencyMaterialConfig.validate(ConfigSaveValidationContext.forChain(config, pipelineConfig));
        ConfigErrors configErrors = dependencyMaterialConfig.errors();
        assertThat(configErrors.isEmpty(), is(false));
        assertThat(configErrors.on(DependencyMaterialConfig.PIPELINE_STAGE_NAME), containsString("Pipeline with name 'pipeline-not-exist' does not exist"));
    }

    @Test
    public void setConfigAttributes_shouldPopulateFromConfigAttributes() {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString(""), new CaseInsensitiveString(""));
        assertThat(dependencyMaterialConfig.getPipelineStageName(), is(nullValue()));
        assertThat(dependencyMaterialConfig.ignoreForScheduling(), is(false));
        HashMap<String, String> configMap = new HashMap<>();
        configMap.put(AbstractMaterialConfig.MATERIAL_NAME, "name1");
        configMap.put(DependencyMaterialConfig.PIPELINE_STAGE_NAME, "pipeline-1 [stage-1]");
        configMap.put(DependencyMaterialConfig.IGNORE_FOR_SCHEDULING, "true");

        dependencyMaterialConfig.setConfigAttributes(configMap);

        assertThat(dependencyMaterialConfig.getMaterialName(), is(new CaseInsensitiveString("name1")));
        assertThat(dependencyMaterialConfig.getPipelineName(), is(new CaseInsensitiveString("pipeline-1")));
        assertThat(dependencyMaterialConfig.getStageName(), is(new CaseInsensitiveString("stage-1")));
        assertThat(dependencyMaterialConfig.getPipelineStageName(), is("pipeline-1 [stage-1]"));
        assertThat(dependencyMaterialConfig.ignoreForScheduling(), is(true));
    }

    @Test
    public void setConfigAttributes_shouldNotPopulateNameFromConfigAttributesIfNameIsEmptyOrNull() {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("name2"), new CaseInsensitiveString("pipeline"), new CaseInsensitiveString("stage"));
        HashMap<String, String> configMap = new HashMap<>();
        configMap.put(AbstractMaterialConfig.MATERIAL_NAME, "");

        dependencyMaterialConfig.setConfigAttributes(configMap);

        assertThat(dependencyMaterialConfig.getMaterialName(), is(nullValue()));
    }

    @Test
    public void shouldValidateTree() {
        DependencyMaterialConfig dependencyMaterialConfig = new DependencyMaterialConfig(new CaseInsensitiveString("upstream_stage"), new CaseInsensitiveString("upstream_pipeline"), new CaseInsensitiveString("stage"));
        PipelineConfig pipeline = new PipelineConfig(new CaseInsensitiveString("p"), new MaterialConfigs());
        pipeline.setOrigin(new FileConfigOrigin());
        dependencyMaterialConfig.validateTree(PipelineConfigSaveValidationContext.forChain(true, "group", config, pipeline));
        assertThat(dependencyMaterialConfig.errors().on(DependencyMaterialConfig.PIPELINE_STAGE_NAME), is("Pipeline with name 'upstream_pipeline' does not exist, it is defined as a dependency for pipeline 'p' (cruise-config.xml)"));
    }
}


text:
var events = require('events')
  , util = require('util')
  , EventEmitter = events.EventEmitter
  , ErrorCodes = require('./ErrorCodes')
  , bufferUtil = require('./BufferUtil').BufferUtil;

/**
 * HyBi Sender implementation
 */

function Sender(socket) {
  this._socket = socket;
  this.firstFragment = true;
}

/**
 * Inherits from EventEmitter.
 */

util.inherits(Sender, events.EventEmitter);

/**
 * Sends a close instruction to the remote party.
 *
 * @api public
 */

Sender.prototype.close = function(code, data, mask) {
  if (typeof code !== 'undefined') {
    if (typeof code !== 'number' ||
      !ErrorCodes.isValidErrorCode(code)) throw new Error('first argument must be a valid error code number');
  }
  code = code || 1000;
  var dataBuffer = new Buffer(2 + (data ? Buffer.byteLength(data) : 0));
  writeUInt16BE.call(dataBuffer, code, 0);
  if (dataBuffer.length > 2) dataBuffer.write(data, 2);
  this.frameAndSend(0x8, dataBuffer, true, mask);
}

/**
 * Sends a ping message to the remote party.
 *
 * @api public
 */

Sender.prototype.ping = function(data, options) {
  var mask = options && options.mask;
  this.frameAndSend(0x9, data || '', true, mask);
}

/**
 * Sends a pong message to the remote party.
 *
 * @api public
 */

Sender.prototype.pong = function(data, options) {
  var mask = options && options.mask;
  this.frameAndSend(0xa, data || '', true, mask);
}

/**
 * Sends text or binary data to the remote party.
 *
 * @api public
 */

Sender.prototype.send = function(data, options, cb) {
  var finalFragment = options && options.fin === false ? false : true;
  var mask = options && options.mask;
  var opcode = options && options.binary ? 2 : 1;
  if (this.firstFragment === false) opcode = 0;
  else this.firstFragment = false;
  if (finalFragment) this.firstFragment = true
  this.frameAndSend(opcode, data, finalFragment, mask, cb);
}

/**
 * Frames and sends a piece of data according to the HyBi WebSocket protocol.
 *
 * @api private
 */

Sender.prototype.frameAndSend = function(opcode, data, finalFragment, maskData, cb) {
  var canModifyData = false;

  if (!data) {
    try {
      this._socket.write(new Buffer([opcode | (finalFragment ? 0x80 : 0), 0]), 'binary', cb);
    }
    catch (e) {
      if (typeof cb == 'function') cb(e);
      else this.emit('error', e);
    }
    return;
  }

  if (!Buffer.isBuffer(data)) {
    canModifyData = true;
    data = (data && typeof data.buffer !== 'undefined') ? getArrayBuffer(data.buffer) : new Buffer(data);
  }

  var dataLength = data.length
    , dataOffset = maskData ? 6 : 2
    , secondByte = dataLength;

  if (dataLength >= 65536) {
    dataOffset += 8;
    secondByte = 127;
  }
  else if (dataLength > 125) {
    dataOffset += 2;
    secondByte = 126;
  }

  var totalLength = (maskData && !canModifyData) ? dataLength + dataOffset : dataOffset;
  var outputBuffer = new Buffer(totalLength);

  outputBuffer[0] = finalFragment ? opcode | 0x80 : opcode;

  switch (secondByte) {
    case 126:
      writeUInt16BE.call(outputBuffer, dataLength, 2);
      break;
    case 127:
      writeUInt32BE.call(outputBuffer, 0, 2);
      writeUInt32BE.call(outputBuffer, dataLength, 6);
  }

  var sendsDone = 0;
  var cbCaller = function() { if (++sendsDone == 2 && typeof cb == 'function') cb(null); }

  if (maskData) {
    outputBuffer[1] = secondByte | 0x80;
    var mask = this._randomMask || (this._randomMask = getRandomMask());
    outputBuffer[dataOffset - 4] = mask[0];
    outputBuffer[dataOffset - 3] = mask[1];
    outputBuffer[dataOffset - 2] = mask[2];
    outputBuffer[dataOffset - 1] = mask[3];
    if (!canModifyData) {
      bufferUtil.mask(data, mask, outputBuffer, dataOffset, dataLength);
      try {
        this._socket.write(outputBuffer, 'binary', cb);
      }
      catch (e) {
        if (typeof cb == 'function') cb(e);
        else this.emit('error', e);
      }
    }
    else {
      bufferUtil.mask(data, mask, data, 0, dataLength);
      try {
        this._socket.write(outputBuffer, 'binary', cb ? cbCaller : null);
        this._socket.write(data, 'binary', cb ? cbCaller : null);
      }
      catch (e) {
        if (typeof cb == 'function') cb(e);
        else this.emit('error', e);
      }
    }
  }
  else {
    outputBuffer[1] = secondByte;
    try {
      this._socket.write(outputBuffer, 'binary', cb ? cbCaller : null);
      this._socket.write(data, 'binary', cb ? cbCaller : null);
    }
    catch (e) {
      if (typeof cb == 'function') cb(e);
      else this.emit('error', e);
    }
  }
}

module.exports = Sender;

function writeUInt16BE(value, offset) {
  this[offset] = (value & 0xff00)>>8;
  this[offset+1] = value & 0xff;
}

function writeUInt32BE(value, offset) {
  this[offset] = (value & 0xff000000)>>24;
  this[offset+1] = (value & 0xff0000)>>16;
  this[offset+2] = (value & 0xff00)>>8;
  this[offset+3] = value & 0xff;
}

function getArrayBuffer(array) {
  var l = array.byteLength
    , buffer = new Buffer(l);
  for (var i = 0; i < l; ++i) {
    buffer[i] = array[i];
  }
  return buffer;
}

function getRandomMask() {
  return new Buffer([
    ~~(Math.random() * 255),
    ~~(Math.random() * 255),
    ~~(Math.random() * 255),
    ~~(Math.random() * 255)
  ]);
}


text:
class Projects::MilestonesController < Projects::ApplicationController
  before_action :module_enabled
  before_action :milestone, only: [:edit, :update, :destroy, :show, :sort_issues, :sort_merge_requests]

  # Allow read any milestone
  before_action :authorize_read_milestone!

  # Allow admin milestone
  before_action :authorize_admin_milestone!, except: [:index, :show]

  respond_to :html

  def index
    @milestones = case params[:state]
                  when 'all'; @project.milestones.order("state, due_date DESC")
                  when 'closed'; @project.milestones.closed.order("due_date DESC")
                  else @project.milestones.active.order("due_date ASC")
                  end

    @milestones = @milestones.includes(:project)
    @milestones = @milestones.page(params[:page]).per(PER_PAGE)
  end

  def new
    @milestone = @project.milestones.new
    respond_with(@milestone)
  end

  def edit
    respond_with(@milestone)
  end

  def show
    @issues = @milestone.issues
    @users = @milestone.participants.uniq
    @merge_requests = @milestone.merge_requests
  end

  def create
    @milestone = Milestones::CreateService.new(project, current_user, milestone_params).execute

    if @milestone.save
      redirect_to namespace_project_milestone_path(@project.namespace,
                                                   @project, @milestone)
    else
      render "new"
    end
  end

  def update
    @milestone = Milestones::UpdateService.new(project, current_user, milestone_params).execute(milestone)

    respond_to do |format|
      format.js
      format.html do
        if @milestone.valid?
          redirect_to namespace_project_milestone_path(@project.namespace,
                                                   @project, @milestone)
        else
          render :edit
        end
      end
    end
  end

  def destroy
    return access_denied! unless can?(current_user, :admin_milestone, @project)

    Milestones::DestroyService.new(project, current_user).execute(milestone)

    respond_to do |format|
      format.html { redirect_to namespace_project_milestones_path }
      format.js { render nothing: true }
    end
  end

  def sort_issues
    @milestone.sort_issues(params['sortable_issue'].map(&:to_i))

    render json: { saved: true }
  end

  def sort_merge_requests
    @merge_requests = @milestone.merge_requests.where(id: params['sortable_merge_request'])
    @merge_requests.each do |merge_request|
      merge_request.position = params['sortable_merge_request'].index(merge_request.id.to_s) + 1
      merge_request.save
    end

    render json: { saved: true }
  end

  protected

  def milestone
    @milestone ||= @project.milestones.find_by!(iid: params[:id])
  end

  def authorize_admin_milestone!
    return render_404 unless can?(current_user, :admin_milestone, @project)
  end

  def module_enabled
    unless @project.issues_enabled || @project.merge_requests_enabled
      return render_404
    end
  end

  def milestone_params
    params.require(:milestone).permit(:title, :description, :due_date, :state_event)
  end
end


text:
package org.hl7.fhir.instance.model;

// Generated on Sat, Aug 22, 2015 23:00-0400 for FHIR v0.5.0

import java.util.*;

import org.hl7.fhir.utilities.Utilities;
import org.hl7.fhir.instance.model.annotations.ResourceDef;
import org.hl7.fhir.instance.model.annotations.SearchParamDefinition;
import org.hl7.fhir.instance.model.annotations.Child;
import org.hl7.fhir.instance.model.annotations.Description;
import org.hl7.fhir.instance.model.annotations.Block;
import org.hl7.fhir.instance.model.api.*;
/**
 * Record details about the anatomical location of a specimen or body part.  This resource may be used when a coded concept does not provide the necessary detail needed for the use case.
 */
@ResourceDef(name="BodySite", profile="http://hl7.org/fhir/Profile/BodySite")
public class BodySite extends DomainResource {

    /**
     * The person to which the body site belongs.
     */
    @Child(name = "patient", type = {Patient.class}, order=0, min=1, max=1, modifier=false, summary=true)
    @Description(shortDefinition="Patient", formalDefinition="The person to which the body site belongs." )
    protected Reference patient;

    /**
     * The actual object that is the target of the reference (The person to which the body site belongs.)
     */
    protected Patient patientTarget;

    /**
     * Identifier for this instance of the anatomical location.
     */
    @Child(name = "identifier", type = {Identifier.class}, order=1, min=0, max=Child.MAX_UNLIMITED, modifier=false, summary=true)
    @Description(shortDefinition="Bodysite identifier", formalDefinition="Identifier for this instance of the anatomical location." )
    protected List<Identifier> identifier;

    /**
     * Named anatomical location - ideally would be coded where possible.
     */
    @Child(name = "code", type = {CodeableConcept.class}, order=2, min=0, max=1, modifier=false, summary=true)
    @Description(shortDefinition="Named anatomical location", formalDefinition="Named anatomical location - ideally would be coded where possible." )
    protected CodeableConcept code;

    /**
     * Modifier to refine the anatomical location.  These include modifiers for laterality, relative location, directionality, number, and plane.
     */
    @Child(name = "modifier", type = {CodeableConcept.class}, order=3, min=0, max=Child.MAX_UNLIMITED, modifier=false, summary=false)
    @Description(shortDefinition="Modification to location code", formalDefinition="Modifier to refine the anatomical location.  These include modifiers for laterality, relative location, directionality, number, and plane." )
    protected List<CodeableConcept> modifier;

    /**
     * Description of anatomical location.
     */
    @Child(name = "description", type = {StringType.class}, order=4, min=0, max=1, modifier=false, summary=true)
    @Description(shortDefinition="The Description of anatomical location", formalDefinition="Description of anatomical location." )
    protected StringType description;

    /**
     * Image or images used to identify a location.
     */
    @Child(name = "image", type = {Attachment.class}, order=5, min=0, max=Child.MAX_UNLIMITED, modifier=false, summary=false)
    @Description(shortDefinition="Attached images", formalDefinition="Image or images used to identify a location." )
    protected List<Attachment> image;

    private static final long serialVersionUID = 1568109920L;

  /*
   * Constructor
   */
    public BodySite() {
      super();
    }

  /*
   * Constructor
   */
    public BodySite(Reference patient) {
      super();
      this.patient = patient;
    }

    /**
     * @return {@link #patient} (The person to which the body site belongs.)
     */
    public Reference getPatient() {
      if (this.patient == null)
        if (Configuration.errorOnAutoCreate())
          throw new Error("Attempt to auto-create BodySite.patient");
        else if (Configuration.doAutoCreate())
          this.patient = new Reference(); // cc
      return this.patient;
    }

    public boolean hasPatient() {
      return this.patient != null && !this.patient.isEmpty();
    }

    /**
     * @param value {@link #patient} (The person to which the body site belongs.)
     */
    public BodySite setPatient(Reference value) {
      this.patient = value;
      return this;
    }

    /**
     * @return {@link #patient} The actual object that is the target of the reference. The reference library doesn't populate this, but you can use it to hold the resource if you resolve it. (The person to which the body site belongs.)
     */
    public Patient getPatientTarget() {
      if (this.patientTarget == null)
        if (Configuration.errorOnAutoCreate())
          throw new Error("Attempt to auto-create BodySite.patient");
        else if (Configuration.doAutoCreate())
          this.patientTarget = new Patient(); // aa
      return this.patientTarget;
    }

    /**
     * @param value {@link #patient} The actual object that is the target of the reference. The reference library doesn't use these, but you can use it to hold the resource if you resolve it. (The person to which the body site belongs.)
     */
    public BodySite setPatientTarget(Patient value) {
      this.patientTarget = value;
      return this;
    }

    /**
     * @return {@link #identifier} (Identifier for this instance of the anatomical location.)
     */
    public List<Identifier> getIdentifier() {
      if (this.identifier == null)
        this.identifier = new ArrayList<Identifier>();
      return this.identifier;
    }

    public boolean hasIdentifier() {
      if (this.identifier == null)
        return false;
      for (Identifier item : this.identifier)
        if (!item.isEmpty())
          return true;
      return false;
    }

    /**
     * @return {@link #identifier} (Identifier for this instance of the anatomical location.)
     */
    // syntactic sugar
    public Identifier addIdentifier() { //3
      Identifier t = new Identifier();
      if (this.identifier == null)
        this.identifier = new ArrayList<Identifier>();
      this.identifier.add(t);
      return t;
    }

    // syntactic sugar
    public BodySite addIdentifier(Identifier t) { //3
      if (t == null)
        return this;
      if (this.identifier == null)
        this.identifier = new ArrayList<Identifier>();
      this.identifier.add(t);
      return this;
    }

    /**
     * @return {@link #code} (Named anatomical location - ideally would be coded where possible.)
     */
    public CodeableConcept getCode() {
      if (this.code == null)
        if (Configuration.errorOnAutoCreate())
          throw new Error("Attempt to auto-create BodySite.code");
        else if (Configuration.doAutoCreate())
          this.code = new CodeableConcept(); // cc
      return this.code;
    }

    public boolean hasCode() {
      return this.code != null && !this.code.isEmpty();
    }

    /**
     * @param value {@link #code} (Named anatomical location - ideally would be coded where possible.)
     */
    public BodySite setCode(CodeableConcept value) {
      this.code = value;
      return this;
    }

    /**
     * @return {@link #modifier} (Modifier to refine the anatomical location.  These include modifiers for laterality, relative location, directionality, number, and plane.)
     */
    public List<CodeableConcept> getModifier() {
      if (this.modifier == null)
        this.modifier = new ArrayList<CodeableConcept>();
      return this.modifier;
    }

    public boolean hasModifier() {
      if (this.modifier == null)
        return false;
      for (CodeableConcept item : this.modifier)
        if (!item.isEmpty())
          return true;
      return false;
    }

    /**
     * @return {@link #modifier} (Modifier to refine the anatomical location.  These include modifiers for laterality, relative location, directionality, number, and plane.)
     */
    // syntactic sugar
    public CodeableConcept addModifier() { //3
      CodeableConcept t = new CodeableConcept();
      if (this.modifier == null)
        this.modifier = new ArrayList<CodeableConcept>();
      this.modifier.add(t);
      return t;
    }

    // syntactic sugar
    public BodySite addModifier(CodeableConcept t) { //3
      if (t == null)
        return this;
      if (this.modifier == null)
        this.modifier = new ArrayList<CodeableConcept>();
      this.modifier.add(t);
      return this;
    }

    /**
     * @return {@link #description} (Description of anatomical location.). This is the underlying object with id, value and extensions. The accessor "getDescription" gives direct access to the value
     */
    public StringType getDescriptionElement() {
      if (this.description == null)
        if (Configuration.errorOnAutoCreate())
          throw new Error("Attempt to auto-create BodySite.description");
        else if (Configuration.doAutoCreate())
          this.description = new StringType(); // bb
      return this.description;
    }

    public boolean hasDescriptionElement() {
      return this.description != null && !this.description.isEmpty();
    }

    public boolean hasDescription() {
      return this.description != null && !this.description.isEmpty();
    }

    /**
     * @param value {@link #description} (Description of anatomical location.). This is the underlying object with id, value and extensions. The accessor "getDescription" gives direct access to the value
     */
    public BodySite setDescriptionElement(StringType value) {
      this.description = value;
      return this;
    }

    /**
     * @return Description of anatomical location.
     */
    public String getDescription() {
      return this.description == null ? null : this.description.getValue();
    }

    /**
     * @param value Description of anatomical location.
     */
    public BodySite setDescription(String value) {
      if (Utilities.noString(value))
        this.description = null;
      else {
        if (this.description == null)
          this.description = new StringType();
        this.description.setValue(value);
      }
      return this;
    }

    /**
     * @return {@link #image} (Image or images used to identify a location.)
     */
    public List<Attachment> getImage() {
      if (this.image == null)
        this.image = new ArrayList<Attachment>();
      return this.image;
    }

    public boolean hasImage() {
      if (this.image == null)
        return false;
      for (Attachment item : this.image)
        if (!item.isEmpty())
          return true;
      return false;
    }

    /**
     * @return {@link #image} (Image or images used to identify a location.)
     */
    // syntactic sugar
    public Attachment addImage() { //3
      Attachment t = new Attachment();
      if (this.image == null)
        this.image = new ArrayList<Attachment>();
      this.image.add(t);
      return t;
    }

    // syntactic sugar
    public BodySite addImage(Attachment t) { //3
      if (t == null)
        return this;
      if (this.image == null)
        this.image = new ArrayList<Attachment>();
      this.image.add(t);
      return this;
    }

      protected void listChildren(List<Property> childrenList) {
        super.listChildren(childrenList);
        childrenList.add(new Property("patient", "Reference(Patient)", "The person to which the body site belongs.", 0, java.lang.Integer.MAX_VALUE, patient));
        childrenList.add(new Property("identifier", "Identifier", "Identifier for this instance of the anatomical location.", 0, java.lang.Integer.MAX_VALUE, identifier));
        childrenList.add(new Property("code", "CodeableConcept", "Named anatomical location - ideally would be coded where possible.", 0, java.lang.Integer.MAX_VALUE, code));
        childrenList.add(new Property("modifier", "CodeableConcept", "Modifier to refine the anatomical location.  These include modifiers for laterality, relative location, directionality, number, and plane.", 0, java.lang.Integer.MAX_VALUE, modifier));
        childrenList.add(new Property("description", "string", "Description of anatomical location.", 0, java.lang.Integer.MAX_VALUE, description));
        childrenList.add(new Property("image", "Attachment", "Image or images used to identify a location.", 0, java.lang.Integer.MAX_VALUE, image));
      }

      public BodySite copy() {
        BodySite dst = new BodySite();
        copyValues(dst);
        dst.patient = patient == null ? null : patient.copy();
        if (identifier != null) {
          dst.identifier = new ArrayList<Identifier>();
          for (Identifier i : identifier)
            dst.identifier.add(i.copy());
        };
        dst.code = code == null ? null : code.copy();
        if (modifier != null) {
          dst.modifier = new ArrayList<CodeableConcept>();
          for (CodeableConcept i : modifier)
            dst.modifier.add(i.copy());
        };
        dst.description = description == null ? null : description.copy();
        if (image != null) {
          dst.image = new ArrayList<Attachment>();
          for (Attachment i : image)
            dst.image.add(i.copy());
        };
        return dst;
      }

      protected BodySite typedCopy() {
        return copy();
      }

      @Override
      public boolean equalsDeep(Base other) {
        if (!super.equalsDeep(other))
          return false;
        if (!(other instanceof BodySite))
          return false;
        BodySite o = (BodySite) other;
        return compareDeep(patient, o.patient, true) && compareDeep(identifier, o.identifier, true) && compareDeep(code, o.code, true)
           && compareDeep(modifier, o.modifier, true) && compareDeep(description, o.description, true) && compareDeep(image, o.image, true)
          ;
      }

      @Override
      public boolean equalsShallow(Base other) {
        if (!super.equalsShallow(other))
          return false;
        if (!(other instanceof BodySite))
          return false;
        BodySite o = (BodySite) other;
        return compareValues(description, o.description, true);
      }

      public boolean isEmpty() {
        return super.isEmpty() && (patient == null || patient.isEmpty()) && (identifier == null || identifier.isEmpty())
           && (code == null || code.isEmpty()) && (modifier == null || modifier.isEmpty()) && (description == null || description.isEmpty())
           && (image == null || image.isEmpty());
      }

  @Override
  public ResourceType getResourceType() {
    return ResourceType.BodySite;
   }

  @SearchParamDefinition(name="code", path="BodySite.code", description="Named anatomical location", type="token" )
  public static final String SP_CODE = "code";
  @SearchParamDefinition(name="patient", path="BodySite.patient", description="Patient to whom bodysite belongs", type="reference" )
  public static final String SP_PATIENT = "patient";

}


text:
<?php
class CrawlerNotConstraintTest extends PHPUnit_Framework_TestCase {

    /**
     * @var Codeception\PHPUnit\Constraint\Crawler
     */
    protected $constraint;

    public function setUp()
    {
        $this->constraint = new Codeception\PHPUnit\Constraint\CrawlerNot('warcraft','/user');
    }

    public function testEvaluation()
    {
        $nodes = new Symfony\Component\DomCrawler\Crawler("
Bye world

Hello world
");
        $this->constraint->evaluate($nodes);
    }

    public function testFailMessageResponse()
    {
        $nodes = new Symfony\Component\DomCrawler\Crawler('
Bye world

Bye warcraft
');
        try {
            $this->constraint->evaluate($nodes->filter('p'), 'selector');
        } catch (PHPUnit_Framework_AssertionFailedError $fail) {
            $this->assertContains("There was 'selector' element on page <bold>/user</bold>", $fail->getMessage());
            $this->assertNotContains('+ <info>
Bye world
</info>',$fail->getMessage());
            $this->assertContains('+ <info>
Bye <bold>warcraft</bold>
</info>',$fail->getMessage());
            return;
        }
        $this->fail("should have failed, but not");
    }

    public function testFailMessageResponseWhenMoreNodes()
    {
        $html = '';
        for ($i = 0; $i < 15; $i++) {
            $html .= "
warcraft $i
";
        }
        $nodes = new Symfony\Component\DomCrawler\Crawler($html);
        try {
            $this->constraint->evaluate($nodes->filter('p'), 'selector');
        } catch (PHPUnit_Framework_AssertionFailedError $fail) {
            $this->assertContains("There was 'selector' element on page <bold>/user</bold>", $fail->getMessage());
            $this->assertContains('+ <info>
<bold>warcraft</bold> 0
</info>',$fail->getMessage());
            $this->assertContains('+ <info>
<bold>warcraft</bold> 14
</info>',$fail->getMessage());
            return;
        }
        $this->fail("should have failed, but not");
    }

    public function testFailMessageResponseWithoutUrl()
    {
        $this->constraint = new Codeception\PHPUnit\Constraint\CrawlerNot('warcraft');
        $nodes = new Symfony\Component\DomCrawler\Crawler('
Bye world

Bye warcraft
');
        try {
            $this->constraint->evaluate($nodes->filter('p'), 'selector');
        } catch (PHPUnit_Framework_AssertionFailedError $fail) {
            $this->assertContains("There was 'selector' element", $fail->getMessage());
            $this->assertNotContains("There was 'selector' element on page <bold>/user</bold>", $fail->getMessage());
            return;
        }
        $this->fail("should have failed, but not");
    }

}