text:
package com.datastax.dse.driver.api.core.graph.remote;

import static com.datastax.dse.driver.api.core.graph.TinkerGraphAssertions.assertThat;
import static com.datastax.dse.driver.internal.core.graph.GraphTestUtils.assertThatContainsLabel;
import static com.datastax.dse.driver.internal.core.graph.GraphTestUtils.assertThatContainsProperties;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;

import com.datastax.dse.driver.Assertions;
import com.datastax.dse.driver.api.core.graph.SocialTraversalSource;
import com.datastax.dse.driver.api.core.graph.TinkerPathAssert;
import com.datastax.dse.driver.api.core.graph.__;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.Version;
import com.datastax.oss.driver.api.core.servererrors.InvalidQueryException;
import com.datastax.oss.driver.api.testinfra.DseRequirement;
import com.datastax.oss.driver.api.testinfra.ccm.CustomCcmRule;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import org.apache.tinkerpop.gremlin.process.traversal.Path;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.Tree;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.assertj.core.api.Assumptions;
import org.junit.Test;

public abstract class GraphTraversalRemoteITBase {

  protected abstract CqlSession session();

  protected abstract boolean isGraphBinary();

  protected abstract GraphTraversalSource graphTraversalSource();

  protected abstract SocialTraversalSource socialTraversalSource();

  protected abstract CustomCcmRule ccmRule();

  /**
   * Ensures that a previously returned {@link Vertex}'s {@link Vertex#id()} can be used as an input
   * to {@link
   * org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource#V(Object...)} to
   * retrieve the {@link Vertex} and that the returned {@link Vertex} is the same.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_use_vertex_id_as_parameter() {
    GraphTraversalSource g = graphTraversalSource();

    // given an existing vertex
    Vertex marko = g.V().hasLabel("person").has("name", "marko").next();
    if (isGraphBinary()) {
      Map<Object, Object> properties =
          g.V().hasLabel("person").has("name", "marko").elementMap("name").next();

      assertThat(properties).containsEntry("name", "marko");
    } else {
      assertThat(marko).hasProperty("name", "marko");
    }

    // then should be able to retrieve that same vertex by id.
    assertThat(g.V(marko.id()).next()).isEqualTo(marko);
  }

  /**
   * Ensures that a previously returned {@link Edge}'s {@link Edge#id()} can be used as an input to
   * {@link
   * org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource#E(Object...)} to
   * retrieve the {@link Edge} and that the returned {@link Edge} is the same.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_use_edge_is_as_parameter() {
    GraphTraversalSource g = graphTraversalSource();

    // given an existing edge
    Edge created = g.E().has("weight", 0.2f).next();

    if (isGraphBinary()) {
      List<Map<Object, Object>> properties =
          g.E().has("weight").elementMap("weight", "software", "person").toList();

      assertThat(properties)
          .anySatisfy(
              props -> {
                assertThatContainsProperties(props, "weight", 0.2f);
                assertThatContainsLabel(props, Direction.IN, "software");
                assertThatContainsLabel(props, Direction.OUT, "person");
              });

    } else {
      assertThat(created)
          .hasProperty("weight", 0.2f)
          .hasInVLabel("software")
          .hasOutVLabel("person");
    }

    // should be able to retrieve incoming and outgoing vertices by edge id
    if (isGraphBinary()) {
      Map<Object, Object> inProperties = g.E(created.id()).inV().elementMap("name", "lang").next();
      Map<Object, Object> outProperties = g.E(created.id()).outV().elementMap("name").next();
      assertThatContainsProperties(inProperties, "name", "lop", "lang", "java");
      assertThatContainsProperties(outProperties, "name", "peter");

    } else {
      Vertex in = g.E(created.id()).inV().next();
      Vertex out = g.E(created.id()).outV().next();

      // should resolve to lop
      assertThat(in).hasLabel("software").hasProperty("name", "lop").hasProperty("lang", "java");

      // should resolve to marko, josh and peter whom created lop.
      assertThat(out).hasLabel("person").hasProperty("name", "peter");
    }
  }

  /**
   * A sanity check that a returned {@link Vertex}'s id is a {@link Map}. This test could break in
   * the future if the format of a vertex ID changes from a Map to something else in DSE.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_deserialize_vertex_id_as_map() {
    GraphTraversalSource g = graphTraversalSource();
    // given an existing vertex
    Vertex marko = g.V().hasLabel("person").has("name", "marko").next();

    // then id should be a map with expected values.
    // Note: this is pretty dependent on DSE Graphs underlying id structure which may vary in
    // the future.
    if (isGraphBinary()) {
      assertThat(((String) marko.id())).contains("marko");
      assertThat(marko.label()).isEqualTo("person");
    } else {
      @SuppressWarnings("unchecked")
      Map<String, String> id = (Map<String, String>) marko.id();
      assertThat(id)
          .hasSize(3)
          .containsEntry("~label", "person")
          .containsKey("community_id")
          .containsKey("member_id");
    }
  }

  /**
   * Ensures that a traversal that returns a result of mixed types is interpreted as a {@link Map}
   * with {@link Object} values. Also uses {@link
   * org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal#by(Traversal)} with an
   * anonymous traversal to get inbound 'created' edges and folds them into a list.
   *
   *
Executes a vertex traversal that binds label 'a' and 'b' to vertex properties and label 'c'
   * to vertices that have edges from that vertex.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_handle_result_object_of_mixed_types() {
    GraphTraversalSource g = graphTraversalSource();
    // find all software vertices and select name, language, and find all vertices that created
    // such software.
    List<Map<String, Object>> results =
        g.V()
            .hasLabel("software")
            .as("a", "b", "c")
            .select("a", "b", "c")
            .by("name")
            .by("lang")
            .by(__.in("created").fold())
            .toList();

    // ensure that lop and ripple and their data are the results return.
    assertThat(results).extracting(m -> m.get("a")).containsOnly("lop", "ripple");

    for (Map<String, Object> result : results) {
      assertThat(result).containsOnlyKeys("a", "b", "c");
      // both software are written in java.
      assertThat(result.get("b")).isEqualTo("java");
      // ensure the created vertices match the creators of the software.
      @SuppressWarnings("unchecked")
      List<Vertex> vertices = (List<Vertex>) result.get("c");
      if (result.get("a").equals("lop")) {
        if (isGraphBinary()) {
          // should contain three vertices
          assertThat(vertices.size()).isEqualTo(3);
        } else {
          // lop, 'c' should contain marko, josh, peter.
          assertThat(vertices)
              .extracting(vertex -> vertex.property("name").value())
              .containsOnly("marko", "josh", "peter");
        }
      } else {
        if (isGraphBinary()) {
          // has only one label
          assertThat(vertices.size()).isEqualTo(1);
        } else {
          assertThat(vertices)
              .extracting(vertex -> vertex.property("name").value())
              .containsOnly("josh");
        }
      }
    }
  }

  /**
   * Ensures that a traversal that returns a sub graph can be retrieved.
   *
   *
The subgraph is all members in a knows relationship, thus is all people who marko knows and
   * the edges that connect them.
   */
  @Test
  public void should_handle_subgraph_graphson() {
    Assumptions.assumeThat(isGraphBinary()).isFalse();

    GraphTraversalSource g = graphTraversalSource();
    // retrieve a subgraph on the knows relationship, this omits the created edges.
    Graph graph = (Graph) g.E().hasLabel("knows").subgraph("subGraph").cap("subGraph").next();

    // there should only be 2 edges (since there are are only 2 knows relationships) and 3 vertices
    assertThat(graph.edges()).toIterable().hasSize(2);
    assertThat(graph.vertices()).toIterable().hasSize(3);
  }

  /**
   * Ensures that a traversal that returns a sub graph can be retrieved.
   *
   *
The subgraph is all members in a knows relationship, thus is all people who marko knows and
   * the edges that connect them.
   */
  @Test
  public void should_handle_subgraph_graph_binary() {
    Assumptions.assumeThat(isGraphBinary()).isTrue();

    GraphTraversalSource g = graphTraversalSource();
    // retrieve a subgraph on the knows relationship, this omits the created edges.
    String graph = (String) g.E().hasLabel("knows").subgraph("subGraph").cap("subGraph").next();

    // there should only be 2 edges (since there are are only 2 knows relationships) and 3 vertices
    assertThat(graph).contains("vertices:3").contains("edges:2");
  }

  /**
   * Ensures a traversal that yields no results is properly retrieved and is empty.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_return_zero_results() {
    if (isGraphBinary()) {
      assertThatThrownBy(() -> graphTraversalSource().V().hasLabel("notALabel").toList())
          .isInstanceOf(InvalidQueryException.class)
          .hasMessageContaining("Unknown vertex label 'notALabel'");
    } else {
      assertThat(graphTraversalSource().V().hasLabel("notALabel").toList()).isEmpty();
    }
  }

  /**
   * Validates that a traversal returning a {@link Tree} structure is returned appropriately with
   * the expected contents.
   *
   *
Retrieves trees of people marko knows and the software they created.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_parse_tree() {
    // Get a tree structure showing the paths from mark to people he knows to software they've
    // created.
    @SuppressWarnings("unchecked")
    Tree<String> tree =
        graphTraversalSource()
            .V()
            .hasLabel("person")
            .out("knows")
            .out("created")
            .tree()
            .by("name")
            .next();

    // Marko knows josh who created lop and ripple.
    assertThat(tree).tree("marko").tree("josh").tree("lop").isLeaf();

    assertThat(tree).tree("marko").tree("josh").tree("ripple").isLeaf();
  }

  /**
   * Validates that a traversal using lambda operations with anonymous traversals are applied
   * appropriately and return the expected results.
   *
   *
Traversal that filters 'person'-labeled vertices by name 'marko' and flatMaps outgoing
   * vertices on the 'knows' relationship by their outgoing 'created' vertices and then maps by
   * their 'name' property and folds them into one list.
   *
   *
Note: This does not validate lambdas with functions as those can't be interpreted and
   * sent remotely.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_handle_lambdas() {
    // Find all people marko knows and the software they created.
    List<Object> software =
        graphTraversalSource()
            .V()
            .hasLabel("person")
            .filter(__.has("name", "marko"))
            .out("knows")
            .flatMap(__.out("created"))
            .map(__.values("name"))
            .fold()
            .next();

    // Marko only knows josh and vadas, of which josh created lop and ripple.
    assertThat(software).containsOnly("lop", "ripple");
  }

  /**
   * Validates that {@link
   * org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal#tryNext()} functions
   * appropriate by returning an {@link Optional} of which the presence of the underlying data
   * depends on whether or not remaining data is present.
   *
   *
This is more of a test of Tinkerpop than the protocol between the client and DSE graph.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_handle_tryNext() {
    GraphTraversal<Vertex, Vertex> traversal =
        graphTraversalSource().V().hasLabel("person").has("name", "marko");

    // value present
    Optional<Vertex> v0 = traversal.tryNext();
    assertThat(v0.isPresent()).isTrue();
    if (!isGraphBinary()) {
      assertThat(v0.get()).hasProperty("name", "marko");
    }

    // value absent as there was only 1 matching vertex.
    Optional<Vertex> v1 = traversal.tryNext();
    assertThat(v1.isPresent()).isFalse();
  }

  /**
   * Validates that {@link GraphTraversal#toStream()} appropriately creates a stream from the
   * underlying iterator on the traversal, and then an attempt to call toStream again yields no
   * results.
   *
   *
This is more of a test of Tinkerpop than the protocol between the client and DSE graph.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_handle_streaming_graphson() {
    Assumptions.assumeThat(isGraphBinary()).isFalse();

    GraphTraversal<Vertex, Vertex> traversal = graphTraversalSource().V().hasLabel("person");
    // retrieve all person vertices to stream, and filter on client side all persons under age 30
    // and map to their name.
    List<String> under30 =
        traversal
            .toStream()
            .filter(v -> v.<Integer>property("age").value() < 30)
            .map(v -> v.<String>property("name").value())
            .collect(Collectors.toList());

    assertThat(under30).containsOnly("marko", "vadas");

    // attempt to get a stream again, which should be empty.
    assertThat(traversal.toStream().collect(Collectors.toList())).isEmpty();
  }

  /**
   * Validates that {@link GraphTraversal#toStream()} appropriately creates a stream from the
   * underlying iterator on the traversal, and then an attempt to call toStream again yields no
   * results.
   *
   *
This is more of a test of Tinkerpop than the protocol between the client and DSE graph.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_handle_streaming_binary() {
    Assumptions.assumeThat(isGraphBinary()).isTrue();

    GraphTraversal<Vertex, Map<Object, Object>> traversal =
        graphTraversalSource().V().hasLabel("person").elementMap("age", "name");
    // retrieve all person vertices to stream, and filter on client side all persons under age 30
    // and map to their name.
    List<String> under30 =
        traversal
            .toStream()
            .filter(v -> (Integer) v.get("age") < 30)
            .map(v -> (String) v.get("name"))
            .collect(Collectors.toList());

    assertThat(under30).containsOnly("marko", "vadas");

    // attempt to get a stream again, which should be empty.
    assertThat(traversal.toStream().collect(Collectors.toList())).isEmpty();
  }

  /**
   * Validates that when traversing a path and labeling some of the elements during the traversal
   * that the output elements are properly labeled.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_resolve_path_with_some_labels() {
    // given a traversal where some objects have labels.
    List<Path> paths =
        graphTraversalSource()
            .V()
            .hasLabel("person")
            .has("name", "marko")
            .as("a")
            .outE("knows")
            .inV()
            .as("c", "d")
            .outE("created")
            .as("e", "f", "g")
            .inV()
            .path()
            .toList();

    // then the paths returned should be labeled for the
    // appropriate objects, and not labeled otherwise.
    for (Path path : paths) {
      TinkerPathAssert.validatePathObjects(path);
      assertThat(path)
          .hasLabel(0, "a")
          .hasNoLabel(1)
          .hasLabel(2, "c", "d")
          .hasLabel(3, "e", "f", "g")
          .hasNoLabel(4);
    }
  }

  /**
   * Validates that when traversing a path and labeling all of the elements during the traversal
   * that the output elements are properly labeled.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_resolve_path_with_labels() {
    // given a traversal where all objects have labels.
    List<Path> paths =
        graphTraversalSource()
            .V()
            .hasLabel("person")
            .has("name", "marko")
            .as("a")
            .outE("knows")
            .as("b")
            .inV()
            .as("c", "d")
            .outE("created")
            .as("e", "f", "g")
            .inV()
            .as("h")
            .path()
            .toList();

    // then the paths returned should be labeled for all
    // objects.
    for (Path path : paths) {
      TinkerPathAssert.validatePathObjects(path);
      Assertions.assertThat(path.labels()).hasSize(5);
      assertThat(path)
          .hasLabel(0, "a")
          .hasLabel(1, "b")
          .hasLabel(2, "c", "d")
          .hasLabel(3, "e", "f", "g")
          .hasLabel(4, "h");
    }
  }

  /**
   * Validates that when traversing a path and labeling none of the elements during the traversal
   * that all the labels are empty in the result.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_resolve_path_without_labels() {
    // given a traversal where no objects have labels.
    List<Path> paths =
        graphTraversalSource()
            .V()
            .hasLabel("person")
            .has("name", "marko")
            .outE("knows")
            .inV()
            .outE("created")
            .inV()
            .path()
            .toList();

    // then the paths returned should be labeled for
    // all objects.
    for (Path path : paths) {
      TinkerPathAssert.validatePathObjects(path);
      for (int i = 0; i < 5; i++) assertThat(path).hasNoLabel(i);
    }
  }

  @Test
  public void should_handle_asynchronous_execution_graphson() {
    Assumptions.assumeThat(isGraphBinary()).isFalse();

    StringBuilder names = new StringBuilder();

    CompletableFuture<List<Vertex>> future =
        graphTraversalSource().V().hasLabel("person").promise(Traversal::toList);
    try {
      // dumb processing to make sure the completable future works correctly and correct results are
      // returned
      future
          .thenAccept(
              vertices -> vertices.forEach(vertex -> names.append((String) vertex.value("name"))))
          .get();
    } catch (InterruptedException | ExecutionException e) {
      fail("Shouldn't have thrown an exception waiting for the result to complete");
    }

    assertThat(names.toString()).contains("peter", "marko", "vadas", "josh");
  }

  @Test
  public void should_handle_asynchronous_execution_graph_binary() {
    Assumptions.assumeThat(isGraphBinary()).isTrue();

    StringBuilder names = new StringBuilder();

    CompletableFuture<List<Vertex>> future =
        graphTraversalSource().V().hasLabel("person").promise(Traversal::toList);
    try {
      // dumb processing to make sure the completable future works correctly and correct results are
      // returned
      future.thenAccept(vertices -> vertices.forEach(vertex -> names.append(vertex.id()))).get();
    } catch (InterruptedException | ExecutionException e) {
      fail("Shouldn't have thrown an exception waiting for the result to complete");
    }

    assertThat(names.toString()).contains("peter", "marko", "vadas", "josh");
  }

  /**
   * Validates that if a traversal is made that encounters an error on the server side that the
   * exception is set on the future.
   *
   * @test_category dse:graph
   */
  @Test
  @DseRequirement(min = "5.1.0")
  public void should_fail_future_returned_from_promise_on_query_error() throws Exception {
    CompletableFuture<?> future =
        graphTraversalSource().V("invalidid").peerPressure().promise(Traversal::next);

    try {
      future.get();
      fail("Expected an ExecutionException");
    } catch (ExecutionException e) {
      assertThat(e.getCause()).isInstanceOf(InvalidQueryException.class);
    }
  }

  /**
   * A simple smoke test to ensure that a user can supply a custom {@link GraphTraversalSource} for
   * use with DSLs.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_allow_use_of_dsl_graphson() {
    Assumptions.assumeThat(isGraphBinary()).isFalse();

    List<Vertex> vertices = socialTraversalSource().persons("marko").knows("vadas").toList();
    assertThat(vertices.size()).isEqualTo(1);
    assertThat(vertices.get(0))
        .hasProperty("name", "marko")
        .hasProperty("age", 29)
        .hasLabel("person");
  }

  /**
   * A simple smoke test to ensure that a user can supply a custom {@link GraphTraversalSource} for
   * use with DSLs.
   *
   * @test_category dse:graph
   */
  @Test
  public void should_allow_use_of_dsl_graph_binary() {
    Assumptions.assumeThat(isGraphBinary()).isTrue();

    List<Map<Object, Object>> vertices =
        socialTraversalSource().persons("marko").knows("vadas").elementMap("name", "age").toList();
    assertThat(vertices.size()).isEqualTo(1);

    assertThatContainsProperties(vertices.get(0), "name", "marko", "age", 29);
    assertThat(vertices.get(0).values()).contains("person");
  }

  /**
   * Ensures that traversals with barriers (which return results bulked) contain the correct amount
   * of end results.
   *
   *
This will fail if ran against DSE < 5.0.9 or DSE < 5.1.2.
   */
  @Test
  public void should_return_correct_results_when_bulked() {
    Optional<Version> dseVersion = ccmRule().getCcmBridge().getDseVersion();
    Assumptions.assumeThat(
            dseVersion.isPresent() && dseVersion.get().compareTo(Version.parse("5.1.2")) > 0)
        .isTrue();

    List<String> results = graphTraversalSource().E().label().barrier().toList();
    Collections.sort(results);

    List<String> expected =
        Arrays.asList("knows", "created", "created", "knows", "created", "created");
    Collections.sort(expected);

    assertThat(results).isEqualTo(expected);
  }
}


text:
package net.dempsy.intern;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.dempsy.Infrastructure;
import net.dempsy.NodeInformation;
import net.dempsy.NodeReceiver;
import net.dempsy.Service;
import net.dempsy.cluster.ClusterInfoException;
import net.dempsy.cluster.ClusterInfoSession;
import net.dempsy.cluster.DirMode;
import net.dempsy.messages.Dispatcher;
import net.dempsy.messages.KeyedMessageWithType;
import net.dempsy.messages.MessageResourceManager;
import net.dempsy.monitoring.NodeStatsCollector;
import net.dempsy.router.RoutingStrategy;
import net.dempsy.router.RoutingStrategy.ContainerAddress;
import net.dempsy.router.RoutingStrategyManager;
import net.dempsy.transport.NodeAddress;
import net.dempsy.transport.RoutedMessage;
import net.dempsy.transport.Sender;
import net.dempsy.transport.TransportManager;
import net.dempsy.util.QuietCloseable;
import net.dempsy.util.SafeString;
import net.dempsy.utils.PersistentTask;

public class OutgoingDispatcher extends Dispatcher implements Service {
    // package protected because it's used in ApplicationState which is really an
    // internal implementation detail of this class.
    static Logger LOGGER = LoggerFactory.getLogger(OutgoingDispatcher.class);
    static Logger LOGGER_SESSION = LoggerFactory.getLogger(OutgoingDispatcher.class.getName() + ".session");
    private static final long RETRY_TIMEOUT = 500L;

    private PersistentTask checkup;
    private final AtomicBoolean isRunning = new AtomicBoolean(false);
    private final RoutingStrategyManager manager;
    private final AtomicReference<ApplicationState> outbounds = new AtomicReference<>(null);

    private final NodeAddress thisNode;
    private final String thisNodeId;
    private final TransportManager tmanager;
    private final NodeReceiver nodeReciever;
    private final AtomicBoolean isReady = new AtomicBoolean(false);
    private final NodeStatsCollector statsCollector;

    public OutgoingDispatcher(final RoutingStrategyManager manager, final NodeAddress thisNode, final String thisNodeId,
        final NodeReceiver nodeReciever, final TransportManager tmanager, final NodeStatsCollector statsCollector) {
        this.manager = manager;
        this.thisNode = thisNode;
        this.thisNodeId = thisNodeId;
        this.tmanager = tmanager;
        this.nodeReciever = nodeReciever;
        this.statsCollector = statsCollector;
    }

    private static class ResourceManagerClosable implements QuietCloseable {
        public final MessageResourceManager disposer;
        public final KeyedMessageWithType toUse;

        public ResourceManagerClosable(final MessageResourceManager disposer, final KeyedMessageWithType message) {
            this.disposer = disposer;
            toUse = disposer != null ? new KeyedMessageWithType(message.key, disposer.replicate(message.message), message.messageTypes) : message;
        }

        @Override
        public void close() {
            if(disposer != null)
                disposer.dispose(toUse.message);
        }
    }

    @Override
    public void dispatch(final KeyedMessageWithType messageParam, final MessageResourceManager disposer) throws InterruptedException {
        final boolean traceEnabled = LOGGER.isTraceEnabled();
        if(messageParam == null)
            throw new NullPointerException("Attempt to dispatch a null message.");

        final Object messageKey = messageParam.key;
        if(messageKey == null)
            throw new NullPointerException("Message " + SafeString.objectDescription(messageParam) + " has a null key.");
        boolean messageSentSomewhere = false;

        try(ResourceManagerClosable x = new ResourceManagerClosable(disposer, messageParam);) {
            final KeyedMessageWithType message = x.toUse;
            ApplicationState tmp = outbounds.get();

            // if we're in the midst of an update then we really want to wait for the new state.
            while(tmp == null) {
                if(!isRunning.get()) {
                    LOGGER.debug("[{}] Router dispatch called while stopped.", thisNodeId);
                    return;
                }

                if(!isReady.get()) // however, if we never were ready then we're not in the midst
                                   // of an update.
                    throw new IllegalStateException("Dispatch used before Router is ready.");

                Thread.yield(); // let the other threads do their thing. Maybe we'll be updated sooner.
                tmp = outbounds.get(); // are we updated yet?
            }
            final ApplicationState cur = tmp;

            final Map<String, RoutingStrategy.Router[]> outboundsByMessageType = cur.outboundsByMessageType;

            // =================================================================================
            // For each message type, determine the set of Routers. The goal of this loop is to set
            // 'containerByNodeAddress'
            final Map<NodeAddress, ContainerAddress> containerByNodeAddress = new HashMap<>();
            for(final String mt: message.messageTypes) {
                final RoutingStrategy.Router[] routers = outboundsByMessageType.get(mt);
                if(routers == null) {
                    if(traceEnabled)
                        LOGGER.trace("[{}] No cluster that handles messages of type {}", thisNodeId, mt);
                } else {
                    // For this message type we now have all of the Routers. For each Router determine
                    // the set of ContainerAddresses that this message will be sent to.
                    for(int i = 0; i < routers.length; i++) {
                        final ContainerAddress ca = routers[i].selectDestinationForMessage(message);
                        // it's possible 'ca' is null when we don't know where to send the message.
                        if(ca == null) {
                            if(LOGGER.isDebugEnabled())
                                LOGGER.debug("[{}] No way to send the message {} to specific cluster for the time being", thisNodeId, message.message);
                        } else {
                            // When the message will be sent to 2 different clusters, but both clusters
                            // are hosted in the same node, then we send 1 message to 1 ContainerAddress
                            // where the 'clusters' field contains both container ids.
                            final ContainerAddress already = containerByNodeAddress.get(ca.node);
                            if(already != null) {
                                final int[] ia = new int[already.clusters.length + ca.clusters.length];
                                System.arraycopy(already.clusters, 0, ia, 0, already.clusters.length);
                                System.arraycopy(ca.clusters, 0, ia, already.clusters.length, ca.clusters.length);
                                containerByNodeAddress.put(ca.node, new ContainerAddress(ca.node, ia));
                            } else
                                containerByNodeAddress.put(ca.node, ca);
                        }
                    }
                }
            }
            // =================================================================================

            if(containerByNodeAddress.size() == 0) {
                if(traceEnabled)
                    LOGGER.trace("[{}] There appears to be no valid destination addresses for the message {}", thisNodeId,
                        SafeString.objectDescription(message.message));
            }

            for(final Map.Entry<NodeAddress, ContainerAddress> e: containerByNodeAddress.entrySet()) {
                final NodeAddress curNode = e.getKey();
                final ContainerAddress curAddr = e.getValue();

                // If we're local then just send this message directly back to our own node.
                if(curNode.equals(thisNode)) {
                    if(traceEnabled)
                        LOGGER.trace("Sending local {}", message);

                    // if the message is a resource then the disposer will be used to dispose of the message
                    // but it needs an additional replicate. See propogateMessageToNode javadoc.
                    nodeReciever.propogateMessageToNode(
                        new RoutedMessage(curAddr.clusters, messageKey,
                            disposer == null ? message.message : disposer.replicate(message.message)),
                        false, // this shouldn't count since Router is an OUTGOING class
                        disposer);
                    messageSentSomewhere = true;
                } else {
                    if(traceEnabled)
                        LOGGER.trace("Sending {} to {}", message, curNode);

                    final Sender sender = cur.getSender(curNode);
                    if(sender == null) {
                        // router update is probably behind the routing strategy update
                        if(isRunning.get())
                            LOGGER.error("[{}] Couldn't send message to " + curNode + " from " + thisNodeId + " because there's no "
                                + Sender.class.getSimpleName(), thisNodeId);
                    } else {
                        sender.send(new RoutedMessage(curAddr.clusters, messageKey,
                            sender.considerMessageOwnsershipTransfered() ? (disposer == null ? message.message : disposer.replicate(message.message))
                                : message.message));
                        messageSentSomewhere = true;
                    }
                }
            }
        } finally {
            if(!messageSentSomewhere) {
                if(traceEnabled)
                    LOGGER.trace("Message not sent.");
                statsCollector.messageNotSent();
            }
        }
    }

    @Override
    public void start(final Infrastructure infra) {
        final ClusterInfoSession session = infra.getCollaborator();
        final String nodesDir = infra.getRootPaths().nodesDir;

        checkup = new PersistentTask(LOGGER_SESSION, isRunning, infra.getScheduler(), RETRY_TIMEOUT) {

            @Override
            public boolean execute() {
                try {
                    // collect up all NodeInfo's known about.
                    session.recursiveMkdir(nodesDir, null, DirMode.PERSISTENT, DirMode.PERSISTENT);
                    final Collection<String> nodeDirs = session.getSubdirs(nodesDir, this);

                    final Set<NodeInformation> alreadySeen = new HashSet<>();
                    // get all of the subdirectories NodeInformations
                    for(final String subdir: nodeDirs) {
                        final NodeInformation ni = (NodeInformation)session.getData(nodesDir + "/" + subdir, null);

                        if(ni == null) {
                            LOGGER_SESSION.warn("[{}] A node directory was empty at " + subdir, thisNodeId);
                            return false;
                        }

                        // see if node info is dupped.
                        if(alreadySeen.contains(ni)) {
                            LOGGER_SESSION.warn("[{}] The node " + ni.nodeAddress + " seems to be registed more than once.", thisNodeId);
                            continue;
                        }

                        if(ni.clusterInfoByClusterId.size() == 0) { // it's ALL adaptor so there's no sense in dealing with it
                            LOGGER_SESSION.trace("[{}] NodeInformation {} appears to be only an Adaptor.", thisNodeId, ni);
                            continue;
                        }

                        alreadySeen.add(ni);
                    }

                    // check to see if there's new nodes.
                    final ApplicationState.Update ud = outbounds.get().update(alreadySeen, thisNode, thisNodeId);

                    if(!ud.change()) {
                        LOGGER_SESSION.info("[{}] Topology change notification resulted in no changes.", thisNodeId);
                        isReady.set(true);
                        return true; // nothing to update.
                    } else if(LOGGER_SESSION.isTraceEnabled())
                        LOGGER_SESSION.info("[{}] Topology change notification resulted in changes ", thisNodeId);

                    // otherwise we will be making changes so remove the current ApplicationState
                    final ApplicationState obs = outbounds.getAndSet(null); // this can cause instability.
                    try {
                        final ApplicationState newState = obs.apply(ud, tmanager, statsCollector, manager, thisNodeId);
                        outbounds.set(newState);
                        isReady.set(true);
                        return true;
                    } catch(final RuntimeException rte) {
                        // if we threw an exception after clearing the outbounds we need to restore it.
                        // This is likely a configuration error so we should probably warn about it.
                        LOGGER_SESSION.warn("[{}] Unexpected exception while applying a topology update", thisNodeId, rte);
                        outbounds.set(obs);
                        throw rte;
                    }
                } catch(final ClusterInfoException e) {
                    final String message = "Failed to find outgoing route information. Will retry shortly.";
                    if(LOGGER_SESSION.isTraceEnabled())
                        LOGGER_SESSION.debug("[{}] {}", new Object[] {thisNodeId,message,e});
                    else
                        LOGGER_SESSION.debug("[{}] {}", thisNodeId, message);
                    return false;
                }
            }

            @Override
            public String toString() {
                return thisNodeId + " find nodes to route to";
            }

        };

        outbounds.set(new ApplicationState(tmanager, thisNode, new ConcurrentHashMap<>()));

        isRunning.set(true);
        checkup.process();
    }

    @Override
    public boolean isReady() {
        if(isReady.get()) {
            final ApplicationState obs = outbounds.get();
            if(obs == null)
                return false;
            if(!manager.isReady()) // this will check the current Routers.
                return false;
            if(!tmanager.isReady())
                return false;

            return true;
        } else
            return false;
    }

    @Override
    public void stop() {
        if(LOGGER.isDebugEnabled())
            LOGGER.debug("{} stopping {}", thisNodeId, OutgoingDispatcher.class.getSimpleName());
        synchronized(isRunning) {
            isRunning.set(false);
            final ApplicationState cur = outbounds.getAndSet(null);
            if(cur != null)
                cur.stop();
        }
    }

    // =====================================================================
    // Strictly for testing.
    // =====================================================================
    public boolean canReach(final String cluterName, final KeyedMessageWithType message) {
        final ApplicationState cur = outbounds.get();
        if(cur == null)
            return false;
        return cur.canReach(cluterName, message);
    }

    public Collection<ContainerAddress> allReachable(final String cluterName) {
        final ApplicationState cur = outbounds.get();
        if(cur == null)
            return new ArrayList<>();
        return cur.allReachable(cluterName);
    }

    public String thisNodeId() {
        return thisNodeId;
    }

    NodeStatsCollector getNodeStatCollector() {
        return statsCollector;
    }
    // =====================================================================

}

redpajama_set_name: RedPajamaGithub


text:
package org.apache.nifi.web.api.entity;

import javax.xml.bind.annotation.XmlRootElement;
import org.apache.nifi.web.api.dto.RevisionDTO;

/**
 * A base type for request/response entities.
 */
@XmlRootElement(name = "entity")
public class Entity {

    private RevisionDTO revision;

    /**
     * A revision for this request/response.
     *
     * @return
     */
    public RevisionDTO getRevision() {
        return revision;
    }

    public void setRevision(RevisionDTO revision) {
        this.revision = revision;
    }

}


text:
package alien4cloud.tosca.parser.postprocess;

import java.util.Map;
import java.util.Set;

import org.alien4cloud.tosca.exceptions.ConstraintValueDoNotMatchPropertyTypeException;
import org.alien4cloud.tosca.exceptions.ConstraintViolationException;
import org.alien4cloud.tosca.model.definitions.PropertyConstraint;
import org.alien4cloud.tosca.model.definitions.PropertyDefinition;
import org.alien4cloud.tosca.model.types.DataType;
import org.alien4cloud.tosca.normative.types.IPropertyType;
import org.alien4cloud.tosca.normative.types.ToscaTypes;
import org.springframework.stereotype.Component;
import org.yaml.snakeyaml.nodes.Node;

import com.google.common.collect.Sets;

import alien4cloud.tosca.context.ToscaContext;
import alien4cloud.tosca.model.ArchiveRoot;
import alien4cloud.tosca.parser.ParsingContextExecution;
import alien4cloud.tosca.parser.ParsingError;
import alien4cloud.tosca.parser.ParsingErrorLevel;
import alien4cloud.tosca.parser.impl.ErrorCode;
import alien4cloud.utils.services.ConstraintPropertyService;

/**
 * Performs validation of a property definition:
 * - check that constraints can apply on type and are well defined
 * - check that default value match the definition.
 */
@Component
public class PropertyDefinitionPostProcessor implements IPostProcessor<Map.Entry<String, PropertyDefinition>> {
    @Override
    public void process(Map.Entry<String, PropertyDefinition> instance) {
        PropertyDefinition propertyDefinition = instance.getValue();
        validateType(propertyDefinition);
        if (propertyDefinition.getConstraints() != null) {
            Set<String> definedConstraints = Sets.newHashSet();
            for (PropertyConstraint constraint : propertyDefinition.getConstraints()) {
                validate(propertyDefinition, constraint);
                if (!definedConstraints.add(constraint.getClass().getName())) {
                    Node node = ParsingContextExecution.getObjectToNodeMap().get(constraint);
                    ParsingContextExecution.getParsingErrors().add(new ParsingError(ParsingErrorLevel.ERROR, ErrorCode.VALIDATION_ERROR,
                            "ToscaPropertyConstraintDuplicate", node.getStartMark(), "Constraint duplicated", node.getEndMark(), "constraint"));
                }
            }
        }

        if (propertyDefinition.getDefault() != null) {
            checkDefaultValue(instance.getKey(), propertyDefinition);
        }
    }

    private void checkDefaultValue(String propertyName, PropertyDefinition propertyDefinition) {
        try {
            ConstraintPropertyService.checkPropertyConstraint(propertyName, propertyDefinition.getDefault().getValue(), propertyDefinition);
        } catch (ConstraintValueDoNotMatchPropertyTypeException | ConstraintViolationException e) {
            Node node = ParsingContextExecution.getObjectToNodeMap().get(propertyDefinition.getDefault());
            StringBuilder problem = new StringBuilder("Validation issue ");
            if (e.getConstraintInformation() != null) {
                problem.append("for " + e.getConstraintInformation().toString());
            }
            problem.append(e.getMessage());
            ParsingContextExecution.getParsingErrors().add(new ParsingError(ParsingErrorLevel.ERROR, ErrorCode.VALIDATION_ERROR, propertyName,
                    node.getStartMark(), problem.toString(), node.getEndMark(), propertyName));
        }
    }

    private void validateType(PropertyDefinition propertyDefinition) {
        String propertyType = propertyDefinition.getType();
        if (propertyType == null) {
            Node node = ParsingContextExecution.getObjectToNodeMap().get(propertyType);
            if (node == null) {
                node = ParsingContextExecution.getObjectToNodeMap().get(propertyDefinition);
            }
            ParsingContextExecution.getParsingErrors().add(new ParsingError(ErrorCode.VALIDATION_ERROR, "ToscaPropertyType", node.getStartMark(),
                    "Property type must be defined", node.getEndMark(), "type"));
        } else if (!ToscaTypes.isSimple(propertyType)) {
            if (ToscaTypes.LIST.equals(propertyType) || ToscaTypes.MAP.equals(propertyType)) {
                PropertyDefinition entrySchema = propertyDefinition.getEntrySchema();
                if (entrySchema == null) {
                    Node node = ParsingContextExecution.getObjectToNodeMap().get(propertyDefinition);
                    ParsingContextExecution.getParsingErrors().add(new ParsingError(ErrorCode.VALIDATION_ERROR, "ToscaPropertyType", node.getStartMark(),
                            "Type " + propertyType + " must define entry schema", node.getEndMark(), "type"));
                } else {
                    validateType(entrySchema);
                }
            } else {
                // It's data type
                ArchiveRoot archiveRoot = ParsingContextExecution.getRootObj();
                if (!archiveRoot.getDataTypes().containsKey(propertyType)) {
                    DataType dataType = ToscaContext.get(DataType.class, propertyType);
                    if (dataType == null) {
                        Node node = ParsingContextExecution.getObjectToNodeMap().get(propertyType);
                        ParsingContextExecution.getParsingErrors().add(new ParsingError(ErrorCode.TYPE_NOT_FOUND, "ToscaPropertyType", node.getStartMark(),
                                "Type " + propertyType + " is not found.", node.getEndMark(), "type"));
                    }
                }
            }
        }
    }

    private void validate(PropertyDefinition propertyDefinition, PropertyConstraint constraint) {
        IPropertyType<?> toscaType = ToscaTypes.fromYamlTypeName(propertyDefinition.getType());
        if (toscaType == null) {
            Node node = ParsingContextExecution.getObjectToNodeMap().get(propertyDefinition.getType());
            ParsingContextExecution.getParsingErrors().add(new ParsingError(ParsingErrorLevel.ERROR, ErrorCode.INVALID_CONSTRAINT, "Constraint parsing issue",
                    node.getStartMark(), "Limitation - Constraint cannot be used for type " + propertyDefinition.getType(), node.getEndMark(), "constraint"));
        } else {
            try {
                constraint.initialize(toscaType);
            } catch (ConstraintValueDoNotMatchPropertyTypeException e) {
                Node node = ParsingContextExecution.getObjectToNodeMap().get(constraint);
                ParsingContextExecution.getParsingErrors().add(new ParsingError(ParsingErrorLevel.ERROR, ErrorCode.VALIDATION_ERROR, "ToscaPropertyConstraint",
                        node.getStartMark(), e.getMessage(), node.getEndMark(), "constraint"));
            }
        }
    }
}


text:
<?xml-stylesheet href="css3-modsel-63.css" type="text/css"?>
<test>

 The color of this text should be green when the pointing device hovers over it.

 The color of this text should be green when the pointing device hovers over it.

</test>


text:
/*global describe, it, beforeEach, expect, jasmine*/
/*eslint camelcase: 0, no-invalid-this: 0 */

const DrawCard = require('../../../server/game/drawcard.js');

describe('DrawCard', function () {
    beforeEach(function () {
        this.testCard = { code: '111', label: 'test 1(some pack)', name: 'test 1' };
        this.gameSpy = jasmine.createSpyObj('game', ['raiseEvent', 'checkWinCondition']);
        this.card = new DrawCard({ game: this.gameSpy }, this.testCard);
    });

    describe('modifyPower()', function() {
        describe('when called with a positive power', function() {
            it('should increase the power on the card and raise a onCardPowerChanged event', function() {
                this.card.modifyPower(2);

                expect(this.card.power).toBe(2);
                expect(this.gameSpy.raiseEvent).toHaveBeenCalledWith('onCardPowerChanged', this.card, 2);
            });
        });

        describe('when called with a negative power', function() {
            it('should reduce the power on the card and raise a onCardPowerChanged event', function() {
                this.card.power = 2;
                this.card.modifyPower(-2);

                expect(this.card.power).toBe(0);
                expect(this.gameSpy.raiseEvent).toHaveBeenCalledWith('onCardPowerChanged', this.card, -2);
            });
        });

        describe('when a card power would go negative', function() {
            it('should set the power of the card to 0', function() {
                this.card.power = 2;
                this.card.modifyPower(-5);

                expect(this.card.power).toBe(0);
                expect(this.gameSpy.raiseEvent).toHaveBeenCalledWith('onCardPowerChanged', this.card, -2);
            });
        });
    });
});


text:
layout: post
title: Brunette Kendra Lust take cock
titleinfo: pornvd
desc: Watch Brunette Kendra Lust take cock. Pornhub is the ultimate xxx porn and sex site.
---

Brunette Kendra Lust take cock

Watch Brunette Kendra Lust take cock. Pornhub is the ultimate xxx porn and sex site.