code: private DataSet<Vertex<K, VV>> createResultSimpleVertex(EdgeDirection messagingDirection,
  		TypeInformation<Tuple2<K, Message>> messageTypeInfo, DataSet<LongValue> numberOfVertices) {
  
  		DataSet<Tuple2<K, Message>> messages;
  
  		TypeInformation<Vertex<K, VV>> vertexTypes = initialVertices.getType();
  
  		final DeltaIteration<Vertex<K, VV>, Vertex<K, VV>> iteration =
  				initialVertices.iterateDelta(initialVertices, this.maximumNumberOfIterations, 0);
  				setUpIteration(iteration);
  
  		switch (messagingDirection) {
  			case IN:
  				messages = buildScatterFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices);
  				break;
  			case OUT:
  				messages = buildScatterFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices);
  				break;
  			case ALL:
  				messages = buildScatterFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices)
  						.union(buildScatterFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices));
  				break;
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  
  		GatherUdf<K, VV, Message> updateUdf = new GatherUdfSimpleVV<>(gatherFunction, vertexTypes);
  
  		// build the update function (co group)
  		CoGroupOperator<?, ?, Vertex<K, VV>> updates =
  				messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
  
  		if (this.configuration != null && this.configuration.isOptNumVertices()) {
  			updates = updates.withBroadcastSet(numberOfVertices, "number of vertices");
  		}
  
  		configureUpdateFunction(updates);
  
  		return iteration.closeWith(updates, updates);
  	}
docs: Creates the operator that represents this scatter-gather graph computation for a simple vertex.
  
  @param messagingDirection
  @param messageTypeInfo
  @param numberOfVertices
  @return the operator