text:
package com.cloudera.dataflow.spark;

import static com.cloudera.dataflow.spark.ShardNameBuilder.getOutputDirectory;
import static com.cloudera.dataflow.spark.ShardNameBuilder.getOutputFilePrefix;
import static com.cloudera.dataflow.spark.ShardNameBuilder.getOutputFileTemplate;
import static com.cloudera.dataflow.spark.ShardNameBuilder.replaceShardCount;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.avro.mapreduce.AvroKeyOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaRDDLike;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

import com.cloudera.dataflow.extension.io.HBaseIO;
import com.cloudera.dataflow.hadoop.HadoopIO;
import com.google.api.client.util.Maps;
import com.google.cloud.dataflow.sdk.coders.CannotProvideCoderException;
import com.google.cloud.dataflow.sdk.coders.Coder;
import com.google.cloud.dataflow.sdk.coders.KvCoder;
import com.google.cloud.dataflow.sdk.io.AvroIO;
import com.google.cloud.dataflow.sdk.io.TextIO;
import com.google.cloud.dataflow.sdk.transforms.Combine;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.transforms.Flatten;
import com.google.cloud.dataflow.sdk.transforms.GroupByKey;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
import com.google.cloud.dataflow.sdk.transforms.ParDo;
import com.google.cloud.dataflow.sdk.transforms.View;
import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
import com.google.cloud.dataflow.sdk.util.WindowedValue;
import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.cloud.dataflow.sdk.values.PCollectionList;
import com.google.cloud.dataflow.sdk.values.PCollectionTuple;
import com.google.cloud.dataflow.sdk.values.PCollectionView;
import com.google.cloud.dataflow.sdk.values.TupleTag;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;

/**
 * Supports translation between a DataFlow transform, and Spark's operations on RDDs.
 */
public final class TransformTranslator {

  private TransformTranslator() {
  }

  private static class FieldGetter {
    private final Map<String, Field> fields;

    FieldGetter(Class<?> clazz) {
      this.fields = Maps.newHashMap();
      for (Field f : clazz.getDeclaredFields()) {
        f.setAccessible(true);
        this.fields.put(f.getName(), f);
      }
    }

    public <T> T get(String fieldname, Object value) {
      try {
        @SuppressWarnings("unchecked")
        T fieldValue = (T) fields.get(fieldname).get(value);
        return fieldValue;
      } catch (IllegalAccessException e) {
        throw new IllegalStateException(e);
      }
    }
  }

  private static <T> TransformEvaluator<Flatten.FlattenPCollectionList<T>> flattenPColl() {
    return new TransformEvaluator<Flatten.FlattenPCollectionList<T>>() {
      @SuppressWarnings("unchecked")
      @Override
      public void evaluate(Flatten.FlattenPCollectionList<T> transform, EvaluationContext context) {
        PCollectionList<T> pcs = context.getInput(transform);
        JavaRDD<T>[] rdds = new JavaRDD[pcs.size()];
        for (int i = 0; i < rdds.length; i++) {
          rdds[i] = (JavaRDD<T>) context.getRDD(pcs.get(i));
        }
        JavaRDD<T> rdd = context.getSparkContext().union(rdds);
        context.setOutputRDD(transform, rdd);
      }
    };
  }

  private static <K, V> TransformEvaluator<GroupByKey.GroupByKeyOnly<K, V>> gbk() {
    return new TransformEvaluator<GroupByKey.GroupByKeyOnly<K, V>>() {
      @Override
      public void evaluate(GroupByKey.GroupByKeyOnly<K, V> transform, EvaluationContext context) {
        @SuppressWarnings("unchecked")
        JavaRDDLike<KV<K, V>, ?> inRDD =
            (JavaRDDLike<KV<K, V>, ?>) context.getInputRDD(transform);
        @SuppressWarnings("unchecked")
        KvCoder<K, V> coder = (KvCoder<K, V>) context.getInput(transform).getCoder();
        final Coder<K> keyCoder = coder.getKeyCoder();
        final Coder<V> valueCoder = coder.getValueCoder();

        // Use coders to convert objects in the PCollection to byte arrays, so they
        // can be transferred over the network for the shuffle.
        JavaRDDLike<KV<K, Iterable<V>>, ?> outRDD = fromPair(toPair(inRDD)
            .mapToPair(CoderHelpers.toByteFunction(keyCoder, valueCoder))
            .groupByKey()
            .mapToPair(CoderHelpers.fromByteFunctionIterable(keyCoder, valueCoder)));
        context.setOutputRDD(transform, outRDD);
      }
    };
  }

  private static final FieldGetter GROUPED_FG = new FieldGetter(Combine.GroupedValues.class);

  private static <K, VI, VO> TransformEvaluator<Combine.GroupedValues<K, VI, VO>> grouped() {
    return new TransformEvaluator<Combine.GroupedValues<K, VI, VO>>() {
      @Override
      public void evaluate(Combine.GroupedValues<K, VI, VO> transform, EvaluationContext context) {
        Combine.KeyedCombineFn<K, VI, ?, VI> keyed = GROUPED_FG.get("fn", transform);
        @SuppressWarnings("unchecked")
        JavaRDDLike<KV<K, Iterable<VI>>, ?> inRDD =
            (JavaRDDLike<KV<K, Iterable<VI>>, ?>) context.getInputRDD(transform);
        context.setOutputRDD(transform, inRDD.map(new KVFunction<>(keyed)));
      }
    };
  }

  private static final FieldGetter COMBINE_GLOBALLY_FG = new FieldGetter(Combine.Globally.class);

  private static <I, A, O> TransformEvaluator<Combine.Globally<I, O>> combineGlobally() {
    return new TransformEvaluator<Combine.Globally<I, O>>() {

      @Override
      public void evaluate(Combine.Globally<I, O> transform, EvaluationContext context) {
        final Combine.CombineFn<I, A, O> globally = COMBINE_GLOBALLY_FG.get("fn", transform);

        @SuppressWarnings("unchecked")
        JavaRDDLike<I, ?> inRdd = (JavaRDDLike<I, ?>) context.getInputRDD(transform);

        final Coder<I> iCoder = context.getInput(transform).getCoder();
        final Coder<A> aCoder;
        try {
          aCoder = globally.getAccumulatorCoder(
              context.getPipeline().getCoderRegistry(), iCoder);
        } catch (CannotProvideCoderException e) {
          throw new IllegalStateException("Could not determine coder for accumulator", e);
        }

        // Use coders to convert objects in the PCollection to byte arrays, so they
        // can be transferred over the network for the shuffle.
        JavaRDD<byte[]> inRddBytes = inRdd.map(CoderHelpers.toByteFunction(iCoder));

        /*A*/ byte[] acc = inRddBytes.aggregate(
            CoderHelpers.toByteArray(globally.createAccumulator(), aCoder),
            new Function2</*A*/ byte[], /*I*/ byte[], /*A*/ byte[]>() {
              @Override
              public /*A*/ byte[] call(/*A*/ byte[] ab, /*I*/ byte[] ib) throws Exception {
                A a = CoderHelpers.fromByteArray(ab, aCoder);
                I i = CoderHelpers.fromByteArray(ib, iCoder);
                return CoderHelpers.toByteArray(globally.addInput(a, i), aCoder);
              }
            },
            new Function2</*A*/ byte[], /*A*/ byte[], /*A*/ byte[]>() {
              @Override
              public /*A*/ byte[] call(/*A*/ byte[] a1b, /*A*/ byte[] a2b) throws Exception {
                A a1 = CoderHelpers.fromByteArray(a1b, aCoder);
                A a2 = CoderHelpers.fromByteArray(a2b, aCoder);
                // don't use Guava's ImmutableList.of as values may be null
                List<A> accumulators = Collections.unmodifiableList(Arrays.asList(a1, a2));
                A merged = globally.mergeAccumulators(accumulators);
                return CoderHelpers.toByteArray(merged, aCoder);
              }
            }
        );
        O output = globally.extractOutput(CoderHelpers.fromByteArray(acc, aCoder));

        Coder<O> coder = context.getOutput(transform).getCoder();
        JavaRDD<byte[]> outRdd = context.getSparkContext().parallelize(
            // don't use Guava's ImmutableList.of as output may be null
            CoderHelpers.toByteArrays(Collections.singleton(output), coder));
        context.setOutputRDD(transform, outRdd.map(CoderHelpers.fromByteFunction(coder)));
      }
    };
  }

  private static final FieldGetter COMBINE_PERKEY_FG = new FieldGetter(Combine.PerKey.class);

  private static <K, VI, VA, VO> TransformEvaluator<Combine.PerKey<K, VI, VO>> combinePerKey() {
    return new TransformEvaluator<Combine.PerKey<K, VI, VO>>() {
      @Override
      public void evaluate(Combine.PerKey<K, VI, VO> transform, EvaluationContext context) {
        final Combine.KeyedCombineFn<K, VI, VA, VO> keyed =
            COMBINE_PERKEY_FG.get("fn", transform);
        @SuppressWarnings("unchecked")
        JavaRDDLike<KV<K, VI>, ?> inRdd =
            (JavaRDDLike<KV<K, VI>, ?>) context.getInputRDD(transform);

        @SuppressWarnings("unchecked")
        KvCoder<K, VI> inputCoder = (KvCoder<K, VI>) context.getInput(transform).getCoder();
        Coder<K> keyCoder = inputCoder.getKeyCoder();
        Coder<VI> viCoder = inputCoder.getValueCoder();
        Coder<VA> vaCoder = null;
        try {
          vaCoder = keyed.getAccumulatorCoder(
              context.getPipeline().getCoderRegistry(), keyCoder, viCoder);
        } catch (CannotProvideCoderException e) {
          throw new IllegalStateException("Could not determine coder for accumulator", e);
        }
        final Coder<KV<K, VI>> kviCoder = KvCoder.of(keyCoder, viCoder);
        final Coder<KV<K, VA>> kvaCoder = KvCoder.of(keyCoder, vaCoder);

        // We need to duplicate K as both the key of the JavaPairRDD as well as inside the value,
        // since the functions passed to combineByKey don't receive the associated key of each
        // value, and we need to map back into methods in Combine.KeyedCombineFn, which each
        // require the key in addition to the VI's and VA's being merged/accumulated. Once Spark
        // provides a way to include keys in the arguments of combine/merge functions, we won't
        // need to duplicate the keys anymore.
        JavaPairRDD<K, KV<K, VI>> inRddDuplicatedKeyPair = inRdd.mapToPair(
            new PairFunction<KV<K, VI>, K, KV<K, VI>>() {
              @Override
              public Tuple2<K, KV<K, VI>> call(KV<K, VI> kv) {
                return new Tuple2<>(kv.getKey(), kv);
              }
            });

        // Use coders to convert objects in the PCollection to byte arrays, so they
        // can be transferred over the network for the shuffle.
        JavaPairRDD<ByteArray, byte[]> inRddDuplicatedKeyPairBytes = inRddDuplicatedKeyPair
            .mapToPair(CoderHelpers.toByteFunction(keyCoder, kviCoder));

        // The output of combineByKey will be "VA" (accumulator) types rather than "VO" (final
        // output types) since Combine.CombineFn only provides ways to merge VAs, and no way
        // to merge VOs.
        JavaPairRDD</*K*/ ByteArray, /*KV<K, VA>*/ byte[]> accumulatedBytes =
            inRddDuplicatedKeyPairBytes.combineByKey(
            new Function</*KV<K, VI>*/ byte[], /*KV<K, VA>*/ byte[]>() {
              @Override
              public /*KV<K, VA>*/ byte[] call(/*KV<K, VI>*/ byte[] input) {
                KV<K, VI> kvi = CoderHelpers.fromByteArray(input, kviCoder);
                VA va = keyed.createAccumulator(kvi.getKey());
                va = keyed.addInput(kvi.getKey(), va, kvi.getValue());
                return CoderHelpers.toByteArray(KV.of(kvi.getKey(), va), kvaCoder);
              }
            },
            new Function2</*KV<K, VA>*/ byte[], /*KV<K, VI>*/ byte[], /*KV<K, VA>*/ byte[]>() {
              @Override
              public /*KV<K, VA>*/ byte[] call(/*KV<K, VA>*/ byte[] acc,
                  /*KV<K, VI>*/ byte[] input) {
                KV<K, VA> kva = CoderHelpers.fromByteArray(acc, kvaCoder);
                KV<K, VI> kvi = CoderHelpers.fromByteArray(input, kviCoder);
                VA va = keyed.addInput(kva.getKey(), kva.getValue(), kvi.getValue());
                kva = KV.of(kva.getKey(), va);
                return CoderHelpers.toByteArray(KV.of(kva.getKey(), kva.getValue()), kvaCoder);
              }
            },
            new Function2</*KV<K, VA>*/ byte[], /*KV<K, VA>*/ byte[], /*KV<K, VA>*/ byte[]>() {
              @Override
              public /*KV<K, VA>*/ byte[] call(/*KV<K, VA>*/ byte[] acc1,
                  /*KV<K, VA>*/ byte[] acc2) {
                KV<K, VA> kva1 = CoderHelpers.fromByteArray(acc1, kvaCoder);
                KV<K, VA> kva2 = CoderHelpers.fromByteArray(acc2, kvaCoder);
                VA va = keyed.mergeAccumulators(kva1.getKey(),
                    // don't use Guava's ImmutableList.of as values may be null
                    Collections.unmodifiableList(Arrays.asList(kva1.getValue(), kva2.getValue())));
                return CoderHelpers.toByteArray(KV.of(kva1.getKey(), va), kvaCoder);
              }
            });

        JavaPairRDD<K, VO> extracted = accumulatedBytes
            .mapToPair(CoderHelpers.fromByteFunction(keyCoder, kvaCoder))
            .mapValues(
                new Function<KV<K, VA>, VO>() {
                  @Override
                  public VO call(KV<K, VA> acc) {
                    return keyed.extractOutput(acc.getKey(), acc.getValue());
                  }
                });
        context.setOutputRDD(transform, fromPair(extracted));
      }
    };
  }

  private static final class KVFunction<K, V> implements Function<KV<K, Iterable<V>>, KV<K, V>> {
    private final Combine.KeyedCombineFn<K, V, ?, V> keyed;

    KVFunction(Combine.KeyedCombineFn<K, V, ?, V> keyed) {
      this.keyed = keyed;
    }

    @Override
    public KV<K, V> call(KV<K, Iterable<V>> kv) throws Exception {
      return KV.of(kv.getKey(), keyed.apply(kv.getKey(), kv.getValue()));
    }
  }

  private static <K, V> JavaPairRDD<K, V> toPair(JavaRDDLike<KV<K, V>, ?> rdd) {
    return rdd.mapToPair(new PairFunction<KV<K, V>, K, V>() {
      @Override
      public Tuple2<K, V> call(KV<K, V> kv) {
        return new Tuple2<>(kv.getKey(), kv.getValue());
      }
    });
  }

  private static <K, V> JavaRDDLike<KV<K, V>, ?> fromPair(JavaPairRDD<K, V> rdd) {
    return rdd.map(new Function<Tuple2<K, V>, KV<K, V>>() {
      @Override
      public KV<K, V> call(Tuple2<K, V> t2) {
        return KV.of(t2._1(), t2._2());
      }
    });
  }

  private static <I, O> TransformEvaluator<ParDo.Bound<I, O>> parDo() {
    return new TransformEvaluator<ParDo.Bound<I, O>>() {
      @Override
      public void evaluate(ParDo.Bound<I, O> transform, EvaluationContext context) {
        DoFnFunction<I, O> dofn =
            new DoFnFunction<>(transform.getFn(),
                context.getRuntimeContext(),
                getSideInputs(transform.getSideInputs(), context));
        @SuppressWarnings("unchecked")
        JavaRDDLike<I, ?> inRDD = (JavaRDDLike<I, ?>) context.getInputRDD(transform);
        context.setOutputRDD(transform, inRDD.mapPartitions(dofn));
      }
    };
  }

  private static final FieldGetter MULTIDO_FG = new FieldGetter(ParDo.BoundMulti.class);

  private static <I, O> TransformEvaluator<ParDo.BoundMulti<I, O>> multiDo() {
    return new TransformEvaluator<ParDo.BoundMulti<I, O>>() {
      @Override
      public void evaluate(ParDo.BoundMulti<I, O> transform, EvaluationContext context) {
        TupleTag<O> mainOutputTag = MULTIDO_FG.get("mainOutputTag", transform);
        MultiDoFnFunction<I, O> multifn = new MultiDoFnFunction<>(
            transform.getFn(),
            context.getRuntimeContext(),
            mainOutputTag,
            getSideInputs(transform.getSideInputs(), context));

        @SuppressWarnings("unchecked")
        JavaRDDLike<I, ?> inRDD = (JavaRDDLike<I, ?>) context.getInputRDD(transform);
        JavaPairRDD<TupleTag<?>, Object> all = inRDD
            .mapPartitionsToPair(multifn)
            .cache();

        PCollectionTuple pct = context.getOutput(transform);
        for (Map.Entry<TupleTag<?>, PCollection<?>> e : pct.getAll().entrySet()) {
          @SuppressWarnings("unchecked")
          JavaPairRDD<TupleTag<?>, Object> filtered =
              all.filter(new TupleTagFilter(e.getKey()));
          context.setRDD(e.getValue(), filtered.values());
        }
      }
    };
  }

  private static <T> TransformEvaluator<TextIO.Read.Bound<T>> readText() {
    return new TransformEvaluator<TextIO.Read.Bound<T>>() {
      @Override
      public void evaluate(TextIO.Read.Bound<T> transform, EvaluationContext context) {
        String pattern = transform.getFilepattern();
        JavaRDD<String> rdd = context.getSparkContext().textFile(pattern);
        context.setOutputRDD(transform, rdd);
      }
    };
  }

  private static <T> TransformEvaluator<TextIO.Write.Bound<T>> writeText() {
    return new TransformEvaluator<TextIO.Write.Bound<T>>() {
      @Override
      public void evaluate(TextIO.Write.Bound<T> transform, EvaluationContext context) {
        @SuppressWarnings("unchecked")
        JavaPairRDD<T, Void> last = ((JavaRDDLike<T, ?>) context.getInputRDD(transform))
            .mapToPair(new PairFunction<T, T,
                Void>() {
              @Override
              public Tuple2<T, Void> call(T t) throws Exception {
                return new Tuple2<>(t, null);
              }
            });
        int shardCount = transform.getNumShards();
        if (shardCount == 0) {
          // use default number of shards, but find the actual number for the template
          shardCount = last.partitions().size();
        } else {
          // number of shards was set explicitly, so repartition
          last = last.repartition(transform.getNumShards());
        }

        String template = replaceShardCount(transform.getShardTemplate(), shardCount);
        String outputDir = getOutputDirectory(transform.getFilenamePrefix(), template);
        String filePrefix = getOutputFilePrefix(transform.getFilenamePrefix(), template);
        String fileTemplate = getOutputFileTemplate(transform.getFilenamePrefix(), template);
        String fileSuffix = transform.getFilenameSuffix();

        Configuration conf = new Configuration();
        conf.set(TemplatedTextOutputFormat.OUTPUT_FILE_PREFIX, filePrefix);
        conf.set(TemplatedTextOutputFormat.OUTPUT_FILE_TEMPLATE, fileTemplate);
        conf.set(TemplatedTextOutputFormat.OUTPUT_FILE_SUFFIX, fileSuffix);
        last.saveAsNewAPIHadoopFile(outputDir, Text.class, NullWritable.class,
            TemplatedTextOutputFormat.class, conf);
      }
    };
  }

  private static <T> TransformEvaluator<AvroIO.Read.Bound<T>> readAvro() {
    return new TransformEvaluator<AvroIO.Read.Bound<T>>() {
      @Override
      public void evaluate(AvroIO.Read.Bound<T> transform, EvaluationContext context) {
        String pattern = transform.getFilepattern();
        JavaSparkContext jsc = context.getSparkContext();
        @SuppressWarnings("unchecked")
        JavaRDD<AvroKey<T>> avroFile = (JavaRDD<AvroKey<T>>) (JavaRDD<?>)
            jsc.newAPIHadoopFile(pattern,
                                 AvroKeyInputFormat.class,
                                 AvroKey.class, NullWritable.class,
                                 new Configuration()).keys();
        JavaRDD<T> rdd = avroFile.map(
            new Function<AvroKey<T>, T>() {
              @Override
              public T call(AvroKey<T> key) {
                return key.datum();
              }
            });
        context.setOutputRDD(transform, rdd);
      }
    };
  }

  private static <T> TransformEvaluator<AvroIO.Write.Bound<T>> writeAvro() {
    return new TransformEvaluator<AvroIO.Write.Bound<T>>() {
      @Override
      public void evaluate(AvroIO.Write.Bound<T> transform, EvaluationContext context) {
        String pattern = transform.getFilenamePrefix();
        @SuppressWarnings("unchecked")
        JavaRDDLike<T, ?> last = (JavaRDDLike<T, ?>) context.getInputRDD(transform);
        Job job;
        try {
          job = Job.getInstance();
        } catch (IOException e) {
          throw new IllegalStateException(e);
        }
        AvroJob.setOutputKeySchema(job, transform.getSchema());
        last.mapToPair(new PairFunction<T, AvroKey<T>, NullWritable>() {
            @Override
            public Tuple2<AvroKey<T>, NullWritable> call(T t) throws Exception {
              return new Tuple2<>(new AvroKey<>(t), NullWritable.get());
            }})
          .saveAsNewAPIHadoopFile(pattern, AvroKey.class, NullWritable.class,
              AvroKeyOutputFormat.class, job.getConfiguration());

      }
    };
  }

  private static <K, V> TransformEvaluator<HadoopIO.Read.Bound<K, V>> readHadoop() {
    return new TransformEvaluator<HadoopIO.Read.Bound<K, V>>() {
      @Override
      public void evaluate(HadoopIO.Read.Bound<K, V> transform, EvaluationContext context) {
        String pattern = transform.getFilepattern();
        JavaSparkContext jsc = context.getSparkContext();

        @SuppressWarnings ("unchecked")
        JavaPairRDD<K, V> file = jsc.newAPIHadoopFile(pattern,
            transform.getFormatClass(),
            transform.getKeyClass(), transform.getValueClass(),
            new Configuration());
        JavaRDD<KV<K, V>> rdd = file.map(new Function<Tuple2<K, V>, KV<K, V>>() {
          @Override
          public KV<K, V> call(Tuple2<K, V> t2) throws Exception {
            return KV.of(t2._1(), t2._2());
          }
        });
        context.setOutputRDD(transform, rdd);
      }
    };
  }

  private static  TransformEvaluator<HBaseIO.Read.Bound>  readHBase () {
     return new TransformEvaluator<HBaseIO.Read.Bound>() {
      /**
		 *
		 */
		private static final long serialVersionUID = 1L;

	@Override
      public void evaluate(HBaseIO.Read.Bound  transform, EvaluationContext context) {
         String tableName = transform.getTableName();
         Configuration conf = HBaseConfiguration.create();
        	    // Add local HBase conf
        //conf.addResource(new Path("file:///opt/mapr/hbase/hbase-0.94.17/conf/hbase-site.xml"))
         conf.set(TableInputFormat.INPUT_TABLE, tableName);
         String[] zkConn = transform.getZkConnection().split(":");
          conf.set("hbase.zookeeper.quorum",zkConn[0]);
         conf.set("hbase.zookeeper.property.clientPort",zkConn[1]);

          JavaSparkContext jsc = context.getSparkContext();

         JavaPairRDD<ImmutableBytesWritable, Result> rdd = jsc.newAPIHadoopRDD(conf, TableInputFormat.class,
        	       org.apache.hadoop.hbase.io.ImmutableBytesWritable.class ,
        	       org.apache.hadoop.hbase.client.Result.class );

        JavaRDD<String[]> rddx = rdd.map(new Function<Tuple2<ImmutableBytesWritable, Result>, String[]>() {
			private static final long serialVersionUID = -5947297712561334213L;

		@Override
          public String[] call(Tuple2<ImmutableBytesWritable, Result> t2) throws Exception {
            String[] res = new String(t2._2.getValue(family, qualifier).getRow()).split(",");
            System.out.println("Result= " + t2._2);
            System.out.println("res= " + res );
            return res ;
          }
        });

        context.setOutputRDD(transform, rddx);
      }
    };
  }

  private static <T> TransformEvaluator<Window.Bound<T>> window() {
    return new TransformEvaluator<Window.Bound<T>>() {
      @Override
      public void evaluate(Window.Bound<T> transform, EvaluationContext context) {
        if (transform.getWindowingStrategy().getWindowFn() instanceof GlobalWindows) {
          context.setOutputRDD(transform, context.getInputRDD(transform));
        } else {
          throw new UnsupportedOperationException("Non-global windowing not supported");
        }
      }
    };
  }

  private static <T> TransformEvaluator<Create<T>> create() {
    return new TransformEvaluator<Create<T>>() {
      @Override
      public void evaluate(Create<T> transform, EvaluationContext context) {
        Iterable<T> elems = transform.getElements();
        // Use a coder to convert the objects in the PCollection to byte arrays, so they
        // can be transferred over the network.
        Coder<T> coder = context.getOutput(transform).getCoder();
        context.setOutputRDDFromValues(transform, elems, coder);
      }
    };
  }

  private static <T> TransformEvaluator<View.AsSingleton<T>> viewAsSingleton() {
    return new TransformEvaluator<View.AsSingleton<T>>() {
      @Override
      public void evaluate(View.AsSingleton<T> transform, EvaluationContext context) {
        Iterable<T> input = context.get(context.getInput(transform));
        context.setPView(context.getOutput(transform), Iterables.transform(input,
            new WindowingFunction<T>()));
      }
    };
  }

  private static <T> TransformEvaluator<View.AsIterable<T>> viewAsIter() {
    return new TransformEvaluator<View.AsIterable<T>>() {
      @Override
      public void evaluate(View.AsIterable<T> transform, EvaluationContext context) {
        Iterable<T> input = context.get(context.getInput(transform));
        context.setPView(context.getOutput(transform), Iterables.transform(input,
            new WindowingFunction<T>()));
      }
    };
  }

  private static <R, W> TransformEvaluator<View.CreatePCollectionView<R, W>> createPCollView() {
    return new TransformEvaluator<View.CreatePCollectionView<R, W>>() {
      @Override
      public void evaluate(View.CreatePCollectionView<R, W> transform, EvaluationContext context) {
        Iterable<WindowedValue<?>> iter = Iterables.transform(
            context.get(context.getInput(transform)), new WindowingFunction<R>());
        context.setPView(context.getOutput(transform), iter);
      }
    };
  }

  private static class WindowingFunction<R> implements com.google.common.base.Function<R,
      WindowedValue<?>> {
    @Override
    public WindowedValue<R> apply(R t) {
      return WindowedValue.valueInGlobalWindow(t);
    }
  }

  private static final class TupleTagFilter<V>
      implements Function<Tuple2<TupleTag<V>, Object>, Boolean> {

    private final TupleTag<V> tag;

    private TupleTagFilter(TupleTag<V> tag) {
      this.tag = tag;
    }

    @Override
    public Boolean call(Tuple2<TupleTag<V>, Object> input) {
      return tag.equals(input._1());
    }
  }

  private static Map<TupleTag<?>, BroadcastHelper<?>> getSideInputs(
      List<PCollectionView<?>> views,
      EvaluationContext context) {
    if (views == null) {
      return ImmutableMap.of();
    } else {
      Map<TupleTag<?>, BroadcastHelper<?>> sideInputs = Maps.newHashMap();
      for (PCollectionView<?> view : views) {
        Iterable<WindowedValue<?>> collectionView = context.getPCollectionView(view);
        Coder<Iterable<WindowedValue<?>>> coderInternal = view.getCoderInternal();
        BroadcastHelper<?> helper = BroadcastHelper.create(collectionView, coderInternal);
        //broadcast side inputs
        helper.broadcast(context.getSparkContext());
        sideInputs.put(view.getTagInternal(), helper);
      }
      return sideInputs;
    }
  }

  private static final Map<Class<? extends PTransform>, TransformEvaluator<?>> EVALUATORS = Maps
      .newHashMap();

  static {
    EVALUATORS.put(TextIO.Read.Bound.class, readText());
    EVALUATORS.put(TextIO.Write.Bound.class, writeText());
    EVALUATORS.put(AvroIO.Read.Bound.class, readAvro());
    EVALUATORS.put(AvroIO.Write.Bound.class, writeAvro());
    EVALUATORS.put(HadoopIO.Read.Bound.class, readHadoop());
    EVALUATORS.put(HBaseIO.Read.Bound.class, readHBase());

    EVALUATORS.put(ParDo.Bound.class, parDo());
    EVALUATORS.put(ParDo.BoundMulti.class, multiDo());
    EVALUATORS.put(GroupByKey.GroupByKeyOnly.class, gbk());
    EVALUATORS.put(Combine.GroupedValues.class, grouped());
    EVALUATORS.put(Combine.Globally.class, combineGlobally());
    EVALUATORS.put(Combine.PerKey.class, combinePerKey());
    EVALUATORS.put(Flatten.FlattenPCollectionList.class, flattenPColl());
    EVALUATORS.put(Create.class, create());
    EVALUATORS.put(View.AsSingleton.class, viewAsSingleton());
    EVALUATORS.put(View.AsIterable.class, viewAsIter());
    EVALUATORS.put(View.CreatePCollectionView.class, createPCollView());
    EVALUATORS.put(Window.Bound.class, window());
  }

  public static <PT extends PTransform<?, ?>> boolean hasTransformEvaluator(Class<PT> clazz) {
    return EVALUATORS.containsKey(clazz);
  }

  public static <PT extends PTransform<?, ?>> TransformEvaluator<PT> getTransformEvaluator(Class<PT>
                                                                                clazz) {
    @SuppressWarnings("unchecked")
    TransformEvaluator<PT> transform = (TransformEvaluator<PT>) EVALUATORS.get(clazz);
    if (transform == null) {
      throw new IllegalStateException("No TransformEvaluator registered for " + clazz);
    }
    return transform;
  }
}


text:
package com.thoughtworks.go.agent;

import com.thoughtworks.go.config.AgentAutoRegistrationProperties;
import org.apache.commons.io.FileUtils;
import org.junit.Rule;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.migrationsupport.rules.EnableRuleMigrationSupport;
import org.junit.rules.TemporaryFolder;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Properties;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.assertj.core.api.Assertions.assertThat;

@EnableRuleMigrationSupport
public class AgentAutoRegistrationPropertiesImplTest {

    @Rule
    public final TemporaryFolder folder = new TemporaryFolder();
    private File configFile;

    @BeforeEach
    void setUp() throws IOException {
        configFile = folder.newFile();
    }

    @Test
    void shouldReturnAgentAutoRegisterPropertiesIfPresent() throws Exception {
        Properties properties = new Properties();

        properties.put(AgentAutoRegistrationPropertiesImpl.AGENT_AUTO_REGISTER_KEY, "foo");
        properties.put(AgentAutoRegistrationPropertiesImpl.AGENT_AUTO_REGISTER_RESOURCES, "foo, zoo");
        properties.put(AgentAutoRegistrationPropertiesImpl.AGENT_AUTO_REGISTER_ENVIRONMENTS, "foo, bar");
        properties.put(AgentAutoRegistrationPropertiesImpl.AGENT_AUTO_REGISTER_HOSTNAME, "agent01.example.com");
        properties.store(new FileOutputStream(configFile), "");

        AgentAutoRegistrationProperties reader = new AgentAutoRegistrationPropertiesImpl(configFile);
        assertThat(reader.agentAutoRegisterKey()).isEqualTo("foo");
        assertThat(reader.agentAutoRegisterResources()).isEqualTo("foo, zoo");
        assertThat(reader.agentAutoRegisterEnvironments()).isEqualTo("foo, bar");
        assertThat(reader.agentAutoRegisterHostname()).isEqualTo("agent01.example.com");
    }

    @Test
    void shouldReturnEmptyStringIfPropertiesNotPresent() {
        AgentAutoRegistrationProperties reader = new AgentAutoRegistrationPropertiesImpl(configFile);
        assertThat(reader.agentAutoRegisterKey().isEmpty()).isTrue();
        assertThat(reader.agentAutoRegisterResources().isEmpty()).isTrue();
        assertThat(reader.agentAutoRegisterEnvironments().isEmpty()).isTrue();
        assertThat(reader.agentAutoRegisterHostname().isEmpty()).isTrue();
    }

    @Test
    void shouldScrubTheAutoRegistrationProperties() throws Exception {
        String originalContents = "" +
                "#\n" +
                "# file autogenerated by chef, any changes will be lost\n" +
                "#\n" +
                "# the registration key\n" +
                "agent.auto.register.key = some secret key\n" +
                "\n" +
                "# the resources on this agent\n" +
                "agent.auto.register.resources = some,resources\n" +
                "\n" +
                "# The hostname of this agent\n" +
                "agent.auto.register.hostname = agent42.example.com\n" +
                "\n" +
                "# The environments this agent belongs to\n" +
                "agent.auto.register.environments = production,blue\n" +
                "\n";
        FileUtils.writeStringToFile(configFile, originalContents, UTF_8);

        AgentAutoRegistrationProperties properties = new AgentAutoRegistrationPropertiesImpl(configFile);
        properties.scrubRegistrationProperties();

        String newContents = "" +
                "#\n" +
                "# file autogenerated by chef, any changes will be lost\n" +
                "#\n" +
                "# the registration key\n" +
                "# The autoregister key has been intentionally removed by Go as a security measure.\n" +
                "# agent.auto.register.key = some secret key\n" +
                "\n" +
                "# the resources on this agent\n" +
                "# This property has been removed by Go after attempting to auto-register with the Go server.\n" +
                "# agent.auto.register.resources = some,resources\n" +
                "\n" +
                "# The hostname of this agent\n" +
                "# This property has been removed by Go after attempting to auto-register with the Go server.\n" +
                "# agent.auto.register.hostname = agent42.example.com\n" +
                "\n" +
                "# The environments this agent belongs to\n" +
                "# This property has been removed by Go after attempting to auto-register with the Go server.\n" +
                "# agent.auto.register.environments = production,blue\n" +
                "\n";
        assertThat(FileUtils.readFileToString(configFile, UTF_8)).isEqualTo(newContents);
    }
}

redpajama_set_name: RedPajamaGithub


text:
<?php

class Kinento_Bankintegration_Model_Settings {
	protected $_options;

	public function toOptionArray() {
		if ( !$this->_options ) {
			$this->getAllOptions();
		}
		return $this->_options;
	}

	public function getAllOptions() {
		if ( !$this->_options ) {
			$this->_options = array();
			$this->_options[] = array( 'value' => false , 'label' => Mage::helper( 'bankintegration' )->__( 'Disabled' ) );
			$this->_options[] = array( 'value' => true , 'label' => Mage::helper( 'bankintegration' )->__( 'Enabled' ) );
		}
		return $this->_options;
	}
}

?>


text:
require File.dirname(__FILE__) + '/../../spec_helper.rb'
require 'starling'

describe Airbrush::Listeners::Memcache, 'initialization' do

  before do
    @host = 'host'
    @frequency = 5
  end

  describe Airbrush::Listeners::Memcache, 'when created' do

    before do
      @memcache = Airbrush::Listeners::Memcache.new(@host, @frequency)
    end

    it 'should configure access to the remote memcache server host' do
      @memcache.host.should == @host
    end

    it 'should configure the poll frequency time' do
      @memcache.poll_frequency.should == @frequency
    end

    # future
    it 'should accept a configurable incoming job queue name'

  end

  describe Airbrush::Listeners::Memcache, 'signal handling' do

    before do
      Signal.stub!(:trap).and_return
    end

    it 'should catch sigint and disconnect from the memcache server' do
      Signal.should_receive(:trap).with('INT').and_return
      Airbrush::Listeners::Memcache.new(@host, @frequency)
    end

  end

  describe Airbrush::Listeners::Memcache, 'when started' do

    before do
      @op = { :command => :command, :args => {}, :id => 'id' }
      @server = mock(MemCache)
      @server.stub!(:get).and_return(@op)
      MemCache.stub!(:new).and_return(@server)

      @handler = mock(Airbrush::Handler)
      @handler.stub!(:process).and_return

      @memcache = Airbrush::Listeners::Memcache.new(@host)
      @memcache.handler = @handler
    end

    it 'should connect to the remote memcache server' # REVISIT: can't call start since it loops forever

    it 'should poll the remote memcache server for jobs' # REVISIT: can't call start since it loops forever

    it 'should sleep 2 seconds inbetween memcache server polls' # REVISIT: can't call start since it loops forever

    it 'should ensure remote jobs are contained inside a hash' do
      @server.stub!(:get).and_return("bad format")
      @handler.should_not_receive(:process)
      @memcache.send :process, @server
    end

    it 'should ensure remote jobs include a :id key' do
      @server.stub!(:get).and_return(@op.except(:id))
      @handler.should_not_receive(:process)
      @memcache.send :process, @server
    end

    it 'should ensure remote jobs include a :command key' do
      @server.stub!(:get).and_return(@op.except(:command))
      @handler.should_not_receive(:process)
      @memcache.send :process, @server
    end

    it 'should ensure remote jobs include an :args key' do
      @server.stub!(:get).and_return(@op.except(:args))
      @handler.should_not_receive(:process)
      @memcache.send :process, @server
    end

    it 'should log invalid jobs' do
      @server.stub!(:get).and_return('invalid job')
      @server.stub!(:valid?).and_return(false)
      @server.log.should_receive(:error).and_return
      @handler.should_not_receive(:process)
      @memcache.send :process, @server
    end

    it 'should ignore invalid jobs (ie. invalid jobs should not halt processing)' do
      @server.stub!(:get).and_return(nil)
      @handler.should_not_receive(:process)

      lambda { @memcache.send :process, @server }.should_not raise_error
    end

    it 'should pass valid jobs to the handler' do
      @handler.should_receive(:process).with(@op[:id], @op[:command], @op[:args]).and_return
      @memcache.send :process, @op
    end

    it 'should log errors return from the handler' do
      @server.stub!(:get).and_return(@op)
      @server.log.should_receive(:error).twice.and_return
      @handler.should_receive(:process).with(@op[:id], @op[:command], @op[:args]).and_raise('Error')
      @memcache.send :process, @op
    end

    it 'should ignore errors returned from the handler (ie. errors raised from the handler should not halt processing)' do
      @server.stub!(:get).and_return(@op)
      @handler.stub!(:process).and_raise("Test Error")

      lambda { @memcache.send :process, @server }.should_not raise_error
    end

    it 'should accept a configurable poll time frequency' do
      @memcache = Airbrush::Listeners::Memcache.new(@host, 10)
      @memcache.poll_frequency.should == 10
    end

    # future
    it 'should automatically discover the target memcache host address via dnssd'
  end

end


text:
my $fold;
my %ARGS;
my %define;

BEGIN {
    BEGIN { unshift @INC, "lib" }
    use Config;
    use strict;

    %ARGS = (CCTYPE => 'MSVC', TARG_DIR => '');

    sub process_cc_flags {
	foreach (map {split /\s+/, $_} @_) {
	    $define{$1} = $2 // 1 if /^-D(\w+)(?:=(.+))?/;
	}
    }

    while (@ARGV) {
	my $flag = shift;
	if ($flag =~ /^(?:CC_FLAGS=)?(-D\w.*)/) {
	    process_cc_flags($1);
	} elsif ($flag =~ /^(CCTYPE|FILETYPE|PLATFORM|TARG_DIR)=(.+)$/) {
	    $ARGS{$1} = $2;
	} elsif ($flag eq '--sort-fold') {
	    ++$fold;
	}
    }
    my @PLATFORM = qw(aix win32 wince os2 netware vms test);
    my %PLATFORM;
    @PLATFORM{@PLATFORM} = ();

    die "PLATFORM undefined, must be one of: @PLATFORM\n"
	unless defined $ARGS{PLATFORM};
    die "PLATFORM must be one of: @PLATFORM\n"
	unless exists $PLATFORM{$ARGS{PLATFORM}};
}
use constant PLATFORM => $ARGS{PLATFORM};

require "./$ARGS{TARG_DIR}regen/embed_lib.pl";

# Is the following guard strictly necessary? Added during refactoring
# to keep the same behaviour when merging other code into here.
process_cc_flags(@Config{qw(ccflags optimize)})
    if $ARGS{PLATFORM} ne 'win32' && $ARGS{PLATFORM} ne 'wince'
    && $ARGS{PLATFORM} ne 'netware';

# Add the compile-time options that miniperl was built with to %define.
# On Win32 these are not the same options as perl itself will be built
# with since miniperl is built with a canned config (one of the win32/
# config_H.*) and none of the BUILDOPT's that are set in the makefiles,
# but they do include some #define's that are hard-coded in various
# source files and header files and don't include any BUILDOPT's that
# the user might have chosen to disable because the canned configs are
# minimal configs that don't include any of those options.

#don't use the host Perl's -V defines for the WinCE Perl
if($ARGS{PLATFORM} ne 'wince') {
    my @options = sort(Config::bincompat_options(), Config::non_bincompat_options());
    print STDERR "Options: (@options)\n" unless $ARGS{PLATFORM} eq 'test';
    $define{$_} = 1 foreach @options;
}

my %exportperlmalloc =
    (
       Perl_malloc		=>	"malloc",
       Perl_mfree		=>	"free",
       Perl_realloc		=>	"realloc",
       Perl_calloc		=>	"calloc",
    );

my $exportperlmalloc = $ARGS{PLATFORM} eq 'os2';

my $config_h = 'config.h';
open(CFG, '<', $config_h) || die "Cannot open $config_h: $!\n";
while (<CFG>) {
    $define{$1} = 1 if /^\s*\#\s*define\s+(MYMALLOC|MULTIPLICITY
                                           |SPRINTF_RETURNS_STRLEN
                                           |KILL_BY_SIGPRC
                                           |(?:PERL|USE|HAS)_\w+)\b/x;
}
close(CFG);

# perl.h logic duplication begins

if ($define{USE_ITHREADS}) {
    if (!$define{MULTIPLICITY}) {
        $define{MULTIPLICITY} = 1;
    }
}

$define{PERL_IMPLICIT_CONTEXT} ||=
    $define{USE_ITHREADS} ||
    $define{MULTIPLICITY} ;

if ($define{USE_ITHREADS} && $ARGS{PLATFORM} ne 'win32' && $^O ne 'darwin') {
    $define{USE_REENTRANT_API} = 1;
}

# perl.h logic duplication ends

print STDERR "Defines: (" . join(' ', sort keys %define) . ")\n"
     unless $ARGS{PLATFORM} eq 'test';

my $sym_ord = 0;
my %ordinal;

if ($ARGS{PLATFORM} eq 'os2') {
    if (open my $fh, '<', 'perl5.def') {
      while (<$fh>) {
	last if /^\s*EXPORTS\b/;
      }
      while (<$fh>) {
	$ordinal{$1} = $2 if /^\s*"(\w+)"\s*(?:=\s*"\w+"\s*)?\@(\d+)\s*$/;
	# This allows skipping ordinals which were used in older versions
	$sym_ord = $1 if /^\s*;\s*LAST_ORDINAL\s*=\s*(\d+)\s*$/;
      }
      $sym_ord < $_ and $sym_ord = $_ for values %ordinal; # Take the max
    }
}

my %skip;
# All platforms export boot_DynaLoader unconditionally.
my %export = ( boot_DynaLoader => 1 );

sub try_symbols {
    foreach my $symbol (@_) {
	++$export{$symbol} unless exists $skip{$symbol};
    }
}

sub readvar {
    # $hash is the hash that we're adding to. For one of our callers, it will
    # actually be the skip hash but that doesn't affect the intent of what
    # we're doing, as in that case we skip adding something to the skip hash
    # for the second time.

    my $file = $ARGS{TARG_DIR} . shift;
    my $hash = shift;
    my $proc = shift;
    open my $vars, '<', $file or die die "Cannot open $file: $!\n";

    while (<$vars>) {
	# All symbols have a Perl_ prefix because that's what embed.h sticks
	# in front of them.  The A?I?S?C? is strictly speaking wrong.
	next unless /\bPERLVAR(A?I?S?C?)\(([IGT]),\s*(\w+)/;

	my $var = "PL_$3";
	my $symbol = $proc ? &$proc($1,$2,$3) : $var;
	++$hash->{$symbol} unless exists $skip{$var};
    }
}

if ($ARGS{PLATFORM} ne 'os2') {
    ++$skip{$_} foreach qw(
		     PL_cryptseen
		     PL_opsave
		     Perl_GetVars
		     Perl_dump_fds
		     Perl_my_bcopy
		     Perl_my_bzero
		     Perl_my_chsize
		     Perl_my_htonl
		     Perl_my_memcmp
		     Perl_my_memset
		     Perl_my_ntohl
		     Perl_my_swap
			 );
    if ($ARGS{PLATFORM} eq 'vms') {
	++$skip{PL_statusvalue_posix};
        # This is a wrapper if we have symlink, not a replacement
        # if we don't.
        ++$skip{Perl_my_symlink} unless $Config{d_symlink};
    } else {
	++$skip{PL_statusvalue_vms};
	++$skip{PL_perllib_sep};
	if ($ARGS{PLATFORM} ne 'aix') {
	    ++$skip{$_} foreach qw(
				PL_DBcv
				PL_generation
				PL_lastgotoprobe
				PL_modcount
				main
				 );
	}
    }
}

if ($ARGS{PLATFORM} ne 'vms') {
    # VMS does its own thing for these symbols.
    ++$skip{$_} foreach qw(
			PL_sig_handlers_initted
			PL_sig_ignoring
			PL_sig_defaulting
			 );
    if ($ARGS{PLATFORM} ne 'win32') {
	++$skip{$_} foreach qw(
			    Perl_do_spawn
			    Perl_do_spawn_nowait
			    Perl_do_aspawn
			     );
    }
}

if ($ARGS{PLATFORM} ne 'win32') {
    ++$skip{$_} foreach qw(
		    Perl_my_setlocale
			 );
}

unless ($define{UNLINK_ALL_VERSIONS}) {
    ++$skip{Perl_unlnk};
}

unless ($define{'DEBUGGING'}) {
    ++$skip{$_} foreach qw(
		    Perl_debop
		    Perl_debprofdump
		    Perl_debstack
		    Perl_debstackptrs
		    Perl_pad_sv
		    Perl_pad_setsv
                    Perl__setlocale_debug_string
		    Perl_set_padlist
		    Perl_hv_assert
		    PL_watchaddr
		    PL_watchok
		    PL_watch_pvx
			 );
}

if ($define{'PERL_IMPLICIT_SYS'}) {
    ++$skip{$_} foreach qw(
		    Perl_my_popen
		    Perl_my_pclose
			 );
    ++$export{$_} foreach qw(perl_get_host_info perl_alloc_override);
    ++$export{perl_clone_host} if $define{USE_ITHREADS};
}
else {
    ++$skip{$_} foreach qw(
		    PL_Mem
		    PL_MemShared
		    PL_MemParse
		    PL_Env
		    PL_StdIO
		    PL_LIO
		    PL_Dir
		    PL_Sock
		    PL_Proc
		    perl_alloc_using
		    perl_clone_using
			 );
}

if (!$define{'PERL_COPY_ON_WRITE'} || $define{'PERL_NO_COW'}) {
    ++$skip{Perl_sv_setsv_cow};
}

unless ($define{PERL_SAWAMPERSAND}) {
    ++$skip{PL_sawampersand};
}

unless ($define{'USE_REENTRANT_API'}) {
    ++$skip{PL_reentrant_buffer};
}

if ($define{'MYMALLOC'}) {
    try_symbols(qw(
		    Perl_dump_mstats
		    Perl_get_mstats
		    Perl_strdup
		    Perl_putenv
		    MallocCfg_ptr
		    MallocCfgP_ptr
		    ));
    unless ($define{USE_ITHREADS}) {
	++$skip{PL_malloc_mutex}
    }
}
else {
    ++$skip{$_} foreach qw(
		    PL_malloc_mutex
		    Perl_dump_mstats
		    Perl_get_mstats
		    MallocCfg_ptr
		    MallocCfgP_ptr
			 );
}

if ($define{'PERL_USE_SAFE_PUTENV'}) {
    ++$skip{PL_use_safe_putenv};
}

unless ($define{'USE_ITHREADS'}) {
    ++$skip{PL_thr_key};
}

# USE_5005THREADS symbols. Kept as reference for easier removal
++$skip{$_} foreach qw(
		    PL_sv_mutex
		    PL_strtab_mutex
		    PL_svref_mutex
		    PL_cred_mutex
		    PL_eval_mutex
		    PL_fdpid_mutex
		    PL_sv_lock_mutex
		    PL_eval_cond
		    PL_eval_owner
		    PL_threads_mutex
		    PL_nthreads
		    PL_nthreads_cond
		    PL_threadnum
		    PL_threadsv_names
		    PL_thrsv
		    PL_vtbl_mutex
		    Perl_condpair_magic
		    Perl_new_struct_thread
		    Perl_per_thread_magicals
		    Perl_thread_create
		    Perl_find_threadsv
		    Perl_unlock_condpair
		    Perl_magic_mutexfree
		    Perl_sv_lock
		     );

unless ($define{'USE_ITHREADS'}) {
    ++$skip{$_} foreach qw(
		    PL_check_mutex
		    PL_op_mutex
		    PL_regex_pad
		    PL_regex_padav
		    PL_dollarzero_mutex
		    PL_hints_mutex
		    PL_locale_mutex
		    PL_my_ctx_mutex
		    PL_perlio_mutex
		    PL_stashpad
		    PL_stashpadix
		    PL_stashpadmax
		    Perl_alloccopstash
		    Perl_allocfilegv
		    Perl_clone_params_del
		    Perl_clone_params_new
		    Perl_parser_dup
		    Perl_dirp_dup
		    Perl_cx_dup
		    Perl_si_dup
		    Perl_any_dup
		    Perl_ss_dup
		    Perl_fp_dup
		    Perl_gp_dup
		    Perl_he_dup
		    Perl_mg_dup
		    Perl_re_dup_guts
		    Perl_sv_dup
		    Perl_sv_dup_inc
		    Perl_rvpv_dup
		    Perl_hek_dup
		    Perl_sys_intern_dup
		    perl_clone
		    perl_clone_using
		    Perl_stashpv_hvname_match
		    Perl_regdupe_internal
		    Perl_newPADOP
			 );
}

unless (   $define{'USE_ITHREADS'}
        && $define{'HAS_NEWLOCALE'})
{
    ++$skip{$_} foreach qw(
        PL_C_locale_obj
    );
}

unless ($define{'PERL_IMPLICIT_CONTEXT'}) {
    ++$skip{$_} foreach qw(
		    PL_my_cxt_index
		    PL_my_cxt_list
		    PL_my_cxt_size
		    PL_my_cxt_keys
		    Perl_croak_nocontext
		    Perl_die_nocontext
		    Perl_deb_nocontext
		    Perl_form_nocontext
		    Perl_load_module_nocontext
		    Perl_mess_nocontext
		    Perl_warn_nocontext
		    Perl_warner_nocontext
		    Perl_newSVpvf_nocontext
		    Perl_sv_catpvf_nocontext
		    Perl_sv_setpvf_nocontext
		    Perl_sv_catpvf_mg_nocontext
		    Perl_sv_setpvf_mg_nocontext
		    Perl_my_cxt_init
		    Perl_my_cxt_index
			 );
}

unless ($define{'PERL_OP_PARENT'}) {
    ++$skip{$_} foreach qw(
		    Perl_op_parent
                );
}

unless ($define{'USE_DTRACE'}) {
    ++$skip{$_} foreach qw(
                    Perl_dtrace_probe_call
                    Perl_dtrace_probe_load
                    Perl_dtrace_probe_op
                    Perl_dtrace_probe_phase
                );
}

unless ($define{'PERL_NEED_APPCTX'}) {
    ++$skip{PL_appctx};
}

unless ($define{'PERL_NEED_TIMESBASE'}) {
    ++$skip{PL_timesbase};
}

unless ($define{'DEBUG_LEAKING_SCALARS'}) {
    ++$skip{PL_sv_serial};
}

unless ($define{'DEBUG_LEAKING_SCALARS_FORK_DUMP'}) {
    ++$skip{PL_dumper_fd};
}

unless ($define{'PERL_DONT_CREATE_GVSV'}) {
    ++$skip{Perl_gv_SVadd};
}

if ($define{'SPRINTF_RETURNS_STRLEN'}) {
    ++$skip{Perl_my_sprintf};
}

unless ($define{'PERL_USES_PL_PIDSTATUS'}) {
    ++$skip{PL_pidstatus};
}

unless ($define{'PERL_TRACK_MEMPOOL'}) {
    ++$skip{PL_memory_debug_header};
}

unless ($define{'MULTIPLICITY'}) {
    ++$skip{$_} foreach qw(
		    PL_interp_size
		    PL_interp_size_5_18_0
			 );
}

unless ($define{'PERL_GLOBAL_STRUCT'}) {
    ++$skip{PL_global_struct_size};
}

unless ($define{'PERL_GLOBAL_STRUCT_PRIVATE'}) {
    ++$skip{$_} foreach qw(
		    PL_my_cxt_keys
		    Perl_my_cxt_index
			 );
}

unless ($define{HAS_MMAP}) {
    ++$skip{PL_mmap_page_size};
}

if ($define{HAS_SIGACTION}) {
    ++$skip{PL_sig_trapped};

    if ($ARGS{PLATFORM} eq 'vms') {
        # FAKE_PERSISTENT_SIGNAL_HANDLERS defined as !defined(HAS_SIGACTION)
        ++$skip{PL_sig_ignoring};
        ++$skip{PL_sig_handlers_initted} unless $define{KILL_BY_SIGPRC};
    }
}

if ($ARGS{PLATFORM} eq 'vms' && !$define{KILL_BY_SIGPRC}) {
    # FAKE_DEFAULT_SIGNAL_HANDLERS defined as KILL_BY_SIGPRC
    ++$skip{Perl_csighandler_init};
    ++$skip{Perl_my_kill};
    ++$skip{Perl_sig_to_vmscondition};
    ++$skip{PL_sig_defaulting};
    ++$skip{PL_sig_handlers_initted} unless !$define{HAS_SIGACTION};
}

unless ($define{USE_LOCALE_COLLATE}) {
    ++$skip{$_} foreach qw(
		    PL_collation_ix
		    PL_collation_name
		    PL_collation_standard
		    PL_collxfrm_base
		    PL_collxfrm_mult
		    Perl_sv_collxfrm
		    Perl_sv_collxfrm_flags
			 );
}

unless ($define{USE_LOCALE_NUMERIC}) {
    ++$skip{$_} foreach qw(
		    PL_numeric_local
		    PL_numeric_name
		    PL_numeric_radix_sv
		    PL_numeric_standard
			 );
}

unless ($define{'USE_C_BACKTRACE'}) {
    ++$skip{Perl_get_c_backtrace_dump};
    ++$skip{Perl_dump_c_backtrace};
}

unless ($define{HAVE_INTERP_INTERN}) {
    ++$skip{$_} foreach qw(
		    Perl_sys_intern_clear
		    Perl_sys_intern_dup
		    Perl_sys_intern_init
		    PL_sys_intern
			 );
}

if ($define{HAS_SIGNBIT}) {
    ++$skip{Perl_signbit};
}

if ($define{'PERL_GLOBAL_STRUCT'}) {
    readvar('perlvars.h', \%skip);
    # This seems like the least ugly way to cope with the fact that PL_sh_path
    # is mentioned in perlvar.h and globvar.sym, and always exported.
    delete $skip{PL_sh_path};
    ++$export{Perl_GetVars};
    try_symbols(qw(PL_Vars PL_VarsPtr))
      unless $ARGS{CCTYPE} eq 'GCC' || $define{PERL_GLOBAL_STRUCT_PRIVATE};
} else {
    ++$skip{$_} foreach qw(Perl_init_global_struct Perl_free_global_struct);
}

++$skip{PL_op_exec_cnt}
    unless $define{PERL_TRACE_OPS};

# functions from *.sym files

my @syms = qw(globvar.sym);

# Symbols that are the public face of the PerlIO layers implementation
# These are in _addition to_ the public face of the abstraction
# and need to be exported to allow XS modules to implement layers
my @layer_syms = qw(
		    PerlIOBase_binmode
		    PerlIOBase_clearerr
		    PerlIOBase_close
		    PerlIOBase_dup
		    PerlIOBase_eof
		    PerlIOBase_error
		    PerlIOBase_fileno
		    PerlIOBase_open
		    PerlIOBase_noop_fail
		    PerlIOBase_noop_ok
		    PerlIOBase_popped
		    PerlIOBase_pushed
		    PerlIOBase_read
		    PerlIOBase_setlinebuf
		    PerlIOBase_unread
		    PerlIOBuf_bufsiz
		    PerlIOBuf_close
		    PerlIOBuf_dup
		    PerlIOBuf_fill
		    PerlIOBuf_flush
		    PerlIOBuf_get_base
		    PerlIOBuf_get_cnt
		    PerlIOBuf_get_ptr
		    PerlIOBuf_open
		    PerlIOBuf_popped
		    PerlIOBuf_pushed
		    PerlIOBuf_read
		    PerlIOBuf_seek
		    PerlIOBuf_set_ptrcnt
		    PerlIOBuf_tell
		    PerlIOBuf_unread
		    PerlIOBuf_write
		    PerlIO_allocate
		    PerlIO_apply_layera
		    PerlIO_apply_layers
		    PerlIO_arg_fetch
		    PerlIO_debug
		    PerlIO_define_layer
		    PerlIO_find_layer
		    PerlIO_isutf8
		    PerlIO_layer_fetch
		    PerlIO_list_alloc
		    PerlIO_list_free
		    PerlIO_modestr
		    PerlIO_parse_layers
		    PerlIO_pending
		    PerlIO_perlio
		    PerlIO_pop
		    PerlIO_push
		    PerlIO_sv_dup
		    Perl_PerlIO_clearerr
		    Perl_PerlIO_close
		    Perl_PerlIO_context_layers
		    Perl_PerlIO_eof
		    Perl_PerlIO_error
		    Perl_PerlIO_fileno
		    Perl_PerlIO_fill
		    Perl_PerlIO_flush
		    Perl_PerlIO_get_base
		    Perl_PerlIO_get_bufsiz
		    Perl_PerlIO_get_cnt
		    Perl_PerlIO_get_ptr
		    Perl_PerlIO_read
		    Perl_PerlIO_restore_errno
		    Perl_PerlIO_save_errno
		    Perl_PerlIO_seek
		    Perl_PerlIO_set_cnt
		    Perl_PerlIO_set_ptrcnt
		    Perl_PerlIO_setlinebuf
		    Perl_PerlIO_stderr
		    Perl_PerlIO_stdin
		    Perl_PerlIO_stdout
		    Perl_PerlIO_tell
		    Perl_PerlIO_unread
		    Perl_PerlIO_write
);
if ($ARGS{PLATFORM} eq 'netware') {
    push(@layer_syms,'PL_def_layerlist','PL_known_layers','PL_perlio');
}

# Export the symbols that make up the PerlIO abstraction, regardless
# of its implementation - read from a file
push @syms, 'perlio.sym';

# PerlIO with layers - export implementation
try_symbols(@layer_syms, 'perlsio_binmode');

unless ($define{'USE_QUADMATH'}) {
  ++$skip{Perl_quadmath_format_needed};
  ++$skip{Perl_quadmath_format_single};
}

###############################################################################

# At this point all skip lists should be completed, as we are about to test
# many symbols against them.

{
    my %seen;
    my ($embed) = setup_embed($ARGS{TARG_DIR});

    foreach (@$embed) {
	my ($flags, $retval, $func, @args) = @$_;
	next unless $func;
	if (   ($flags =~ /[AX]/ && $flags !~ /[xmi]/)
            || ($flags =~ /b/ && ! $define{'NO_MATHOMS'}))
        {
	    # public API, so export

	    # If a function is defined twice, for example before and after
	    # an #else, only export its name once. Important to do this test
	    # within the block, as the *first* definition may have flags which
	    # mean "don't export"
	    next if $seen{$func}++;
	    # Should we also skip adding the Perl_ prefix if $flags =~ /o/ ?
	    $func = "Perl_$func" if ($flags =~ /[pX]/ && $func !~ /^Perl_/);
	    ++$export{$func} unless exists $skip{$func};
	}
    }
}

foreach (@syms) {
    my $syms = $ARGS{TARG_DIR} . $_;
    open my $global, '<', $syms or die "failed to open $syms: $!\n";
    while (<$global>) {
	next unless /^([A-Za-z].*)/;
	my $symbol = "$1";
	++$export{$symbol} unless exists $skip{$symbol};
    }
}

# variables

if ($define{'MULTIPLICITY'} && $define{PERL_GLOBAL_STRUCT}) {
    readvar('perlvars.h', \%export, sub { "Perl_" . $_[1] . $_[2] . "_ptr" });
    # XXX AIX seems to want the perlvars.h symbols, for some reason
    if ($ARGS{PLATFORM} eq 'aix' or $ARGS{PLATFORM} eq 'os2') {	# OS/2 needs PL_thr_key
	readvar('perlvars.h', \%export);
    }
}
else {
    unless ($define{'PERL_GLOBAL_STRUCT'}) {
	readvar('perlvars.h', \%export);
    }
    unless ($define{MULTIPLICITY}) {
	readvar('intrpvar.h', \%export);
    }
}

# Oddities from PerlIO
# All have alternate implementations in perlio.c, so always exist.
# Should they be considered to be part of the API?
try_symbols(qw(
		    PerlIO_binmode
		    PerlIO_getpos
		    PerlIO_init
		    PerlIO_setpos
		    PerlIO_tmpfile
	     ));

if ($ARGS{PLATFORM} eq 'win32') {
    try_symbols(qw(
				 win32_free_childdir
				 win32_free_childenv
				 win32_get_childdir
				 win32_get_childenv
				 win32_spawnvp
		 ));
}

if ($ARGS{PLATFORM} eq 'wince') {
    ++$skip{'win32_isatty'}; # commit 4342f4d6df is win32-only
}

if ($ARGS{PLATFORM} =~ /^win(?:32|ce)$/) {
    try_symbols(qw(
			    Perl_init_os_extras
			    Perl_thread_create
			    Perl_win32_init
			    Perl_win32_term
			    RunPerl
			    win32_async_check
			    win32_errno
			    win32_environ
			    win32_abort
			    win32_fstat
			    win32_stat
			    win32_pipe
			    win32_popen
			    win32_pclose
			    win32_rename
			    win32_setmode
			    win32_chsize
			    win32_lseek
			    win32_tell
			    win32_dup
			    win32_dup2
			    win32_open
			    win32_close
			    win32_eof
			    win32_isatty
			    win32_read
			    win32_write
			    win32_mkdir
			    win32_rmdir
			    win32_chdir
			    win32_flock
			    win32_execv
			    win32_execvp
			    win32_htons
			    win32_ntohs
			    win32_htonl
			    win32_ntohl
			    win32_inet_addr
			    win32_inet_ntoa
			    win32_socket
			    win32_bind
			    win32_listen
			    win32_accept
			    win32_connect
			    win32_send
			    win32_sendto
			    win32_recv
			    win32_recvfrom
			    win32_shutdown
			    win32_closesocket
			    win32_ioctlsocket
			    win32_setsockopt
			    win32_getsockopt
			    win32_getpeername
			    win32_getsockname
			    win32_gethostname
			    win32_gethostbyname
			    win32_gethostbyaddr
			    win32_getprotobyname
			    win32_getprotobynumber
			    win32_getservbyname
			    win32_getservbyport
			    win32_select
			    win32_endhostent
			    win32_endnetent
			    win32_endprotoent
			    win32_endservent
			    win32_getnetent
			    win32_getnetbyname
			    win32_getnetbyaddr
			    win32_getprotoent
			    win32_getservent
			    win32_sethostent
			    win32_setnetent
			    win32_setprotoent
			    win32_setservent
			    win32_getenv
			    win32_putenv
			    win32_perror
			    win32_malloc
			    win32_calloc
			    win32_realloc
			    win32_free
			    win32_sleep
			    win32_times
			    win32_access
			    win32_alarm
			    win32_chmod
			    win32_open_osfhandle
			    win32_get_osfhandle
			    win32_ioctl
			    win32_link
			    win32_unlink
			    win32_utime
			    win32_gettimeofday
			    win32_uname
			    win32_wait
			    win32_waitpid
			    win32_kill
			    win32_str_os_error
			    win32_opendir
			    win32_readdir
			    win32_telldir
			    win32_seekdir
			    win32_rewinddir
			    win32_closedir
			    win32_longpath
			    win32_ansipath
			    win32_os_id
			    win32_getpid
			    win32_crypt
			    win32_dynaload
			    win32_clearenv
			    win32_stdin
			    win32_stdout
			    win32_stderr
			    win32_ferror
			    win32_feof
			    win32_strerror
			    win32_fprintf
			    win32_printf
			    win32_vfprintf
			    win32_vprintf
			    win32_fread
			    win32_fwrite
			    win32_fopen
			    win32_fdopen
			    win32_freopen
			    win32_fclose
			    win32_fputs
			    win32_fputc
			    win32_ungetc
			    win32_getc
			    win32_fileno
			    win32_clearerr
			    win32_fflush
			    win32_ftell
			    win32_fseek
			    win32_fgetpos
			    win32_fsetpos
			    win32_rewind
			    win32_tmpfile
			    win32_setbuf
			    win32_setvbuf
			    win32_flushall
			    win32_fcloseall
			    win32_fgets
			    win32_gets
			    win32_fgetc
			    win32_putc
			    win32_puts
			    win32_getchar
			    win32_putchar
		 ));
}
elsif ($ARGS{PLATFORM} eq 'vms') {
    try_symbols(qw(
		      Perl_cando
		      Perl_cando_by_name
		      Perl_closedir
		      Perl_csighandler_init
		      Perl_do_rmdir
		      Perl_fileify_dirspec
		      Perl_fileify_dirspec_ts
		      Perl_fileify_dirspec_utf8
		      Perl_fileify_dirspec_utf8_ts
		      Perl_flex_fstat
		      Perl_flex_lstat
		      Perl_flex_stat
		      Perl_kill_file
		      Perl_my_chdir
		      Perl_my_chmod
		      Perl_my_crypt
		      Perl_my_endpwent
		      Perl_my_fclose
		      Perl_my_fdopen
		      Perl_my_fgetname
		      Perl_my_flush
		      Perl_my_fwrite
		      Perl_my_gconvert
		      Perl_my_getenv
		      Perl_my_getenv_len
		      Perl_my_getpwnam
		      Perl_my_getpwuid
		      Perl_my_gmtime
		      Perl_my_kill
		      Perl_my_killpg
		      Perl_my_localtime
		      Perl_my_mkdir
		      Perl_my_sigaction
		      Perl_my_symlink
		      Perl_my_time
		      Perl_my_tmpfile
		      Perl_my_trnlnm
		      Perl_my_utime
		      Perl_my_waitpid
		      Perl_opendir
		      Perl_pathify_dirspec
		      Perl_pathify_dirspec_ts
		      Perl_pathify_dirspec_utf8
		      Perl_pathify_dirspec_utf8_ts
		      Perl_readdir
		      Perl_readdir_r
		      Perl_rename
		      Perl_rmscopy
		      Perl_rmsexpand
		      Perl_rmsexpand_ts
		      Perl_rmsexpand_utf8
		      Perl_rmsexpand_utf8_ts
		      Perl_seekdir
		      Perl_sig_to_vmscondition
		      Perl_telldir
		      Perl_tounixpath
		      Perl_tounixpath_ts
		      Perl_tounixpath_utf8
		      Perl_tounixpath_utf8_ts
		      Perl_tounixspec
		      Perl_tounixspec_ts
		      Perl_tounixspec_utf8
		      Perl_tounixspec_utf8_ts
		      Perl_tovmspath
		      Perl_tovmspath_ts
		      Perl_tovmspath_utf8
		      Perl_tovmspath_utf8_ts
		      Perl_tovmsspec
		      Perl_tovmsspec_ts
		      Perl_tovmsspec_utf8
		      Perl_tovmsspec_utf8_ts
		      Perl_trim_unixpath
		      Perl_vms_case_tolerant
		      Perl_vms_do_aexec
		      Perl_vms_do_exec
		      Perl_vms_image_init
		      Perl_vms_realpath
		      Perl_vmssetenv
		      Perl_vmssetuserlnm
		      Perl_vmstrnenv
		      PerlIO_openn
		 ));
}
elsif ($ARGS{PLATFORM} eq 'os2') {
    try_symbols(qw(
		      ctermid
		      get_sysinfo
		      Perl_OS2_init
		      Perl_OS2_init3
		      Perl_OS2_term
		      OS2_Perl_data
		      dlopen
		      dlsym
		      dlerror
		      dlclose
		      dup2
		      dup
		      my_tmpfile
		      my_tmpnam
		      my_flock
		      my_rmdir
		      my_mkdir
		      my_getpwuid
		      my_getpwnam
		      my_getpwent
		      my_setpwent
		      my_endpwent
		      fork_with_resources
		      croak_with_os2error
		      setgrent
		      endgrent
		      getgrent
		      malloc_mutex
		      threads_mutex
		      nthreads
		      nthreads_cond
		      os2_cond_wait
		      os2_stat
		      os2_execname
		      async_mssleep
		      msCounter
		      InfoTable
		      pthread_join
		      pthread_create
		      pthread_detach
		      XS_Cwd_change_drive
		      XS_Cwd_current_drive
		      XS_Cwd_extLibpath
		      XS_Cwd_extLibpath_set
		      XS_Cwd_sys_abspath
		      XS_Cwd_sys_chdir
		      XS_Cwd_sys_cwd
		      XS_Cwd_sys_is_absolute
		      XS_Cwd_sys_is_relative
		      XS_Cwd_sys_is_rooted
		      XS_DynaLoader_mod2fname
		      XS_File__Copy_syscopy
		      Perl_Register_MQ
		      Perl_Deregister_MQ
		      Perl_Serve_Messages
		      Perl_Process_Messages
		      init_PMWIN_entries
		      PMWIN_entries
		      Perl_hab_GET
		      loadByOrdinal
		      pExtFCN
		      os2error
		      ResetWinError
		      CroakWinError
		      PL_do_undump
		 ));
}
elsif ($ARGS{PLATFORM} eq 'netware') {
    try_symbols(qw(
			Perl_init_os_extras
			Perl_thread_create
			Perl_nw5_init
			RunPerl
			AllocStdPerl
			FreeStdPerl
			do_spawn2
			do_aspawn
			nw_uname
			nw_stdin
			nw_stdout
			nw_stderr
			nw_feof
			nw_ferror
			nw_fopen
			nw_fclose
			nw_clearerr
			nw_getc
			nw_fgets
			nw_fputc
			nw_fputs
			nw_fflush
			nw_ungetc
			nw_fileno
			nw_fdopen
			nw_freopen
			nw_fread
			nw_fwrite
			nw_setbuf
			nw_setvbuf
			nw_vfprintf
			nw_ftell
			nw_fseek
			nw_rewind
			nw_tmpfile
			nw_fgetpos
			nw_fsetpos
			nw_dup
			nw_access
			nw_chmod
			nw_chsize
			nw_close
			nw_dup2
			nw_flock
			nw_isatty
			nw_link
			nw_lseek
			nw_stat
			nw_mktemp
			nw_open
			nw_read
			nw_rename
			nw_setmode
			nw_unlink
			nw_utime
			nw_write
			nw_chdir
			nw_rmdir
			nw_closedir
			nw_opendir
			nw_readdir
			nw_rewinddir
			nw_seekdir
			nw_telldir
			nw_htonl
			nw_htons
			nw_ntohl
			nw_ntohs
			nw_accept
			nw_bind
			nw_connect
			nw_endhostent
			nw_endnetent
			nw_endprotoent
			nw_endservent
			nw_gethostbyaddr
			nw_gethostbyname
			nw_gethostent
			nw_gethostname
			nw_getnetbyaddr
			nw_getnetbyname
			nw_getnetent
			nw_getpeername
			nw_getprotobyname
			nw_getprotobynumber
			nw_getprotoent
			nw_getservbyname
			nw_getservbyport
			nw_getservent
			nw_getsockname
			nw_getsockopt
			nw_inet_addr
			nw_listen
			nw_socket
			nw_recv
			nw_recvfrom
			nw_select
			nw_send
			nw_sendto
			nw_sethostent
			nw_setnetent
			nw_setprotoent
			nw_setservent
			nw_setsockopt
			nw_inet_ntoa
			nw_shutdown
			nw_crypt
			nw_execvp
			nw_kill
			nw_Popen
			nw_Pclose
			nw_Pipe
			nw_times
			nw_waitpid
			nw_getpid
			nw_spawnvp
			nw_os_id
			nw_open_osfhandle
			nw_get_osfhandle
			nw_abort
			nw_sleep
			nw_wait
			nw_dynaload
			nw_strerror
			fnFpSetMode
			fnInsertHashListAddrs
			fnGetHashListAddrs
			Perl_deb
			Perl_sv_setsv
			Perl_sv_catsv
			Perl_sv_catpvn
			Perl_sv_2pv
			nw_freeenviron
			Remove_Thread_Ctx
		 ));
}

# When added this code was only run for Win32 and WinCE
# Currently only Win32 links static extensions into the shared library.
# The WinCE makefile doesn't appear to support static extensions, so this code
# can't have any effect there.
# The NetWare Makefile doesn't support static extensions (and hardcodes the
# list of dynamic extensions, and the rules to build them)
# For *nix (and presumably OS/2) with a shared libperl, Makefile.SH compiles
# static extensions with -fPIC, but links them to perl, not libperl.so
# The VMS build scripts don't yet implement static extensions at all.

if ($ARGS{PLATFORM} =~ /^win(?:32|ce)$/) {
    # records of type boot_module for statically linked modules (except Dynaloader)
    my $static_ext = $Config{static_ext} // "";
    $static_ext =~ s/\//__/g;
    $static_ext =~ s/\bDynaLoader\b//;
    try_symbols(map {"boot_$_"} grep {/\S/} split /\s+/, $static_ext);
    try_symbols("init_Win32CORE") if $static_ext =~ /\bWin32CORE\b/;
}

if ($ARGS{PLATFORM} eq 'os2') {
    my (%mapped, @missing);
    open MAP, '<', 'miniperl.map' or die 'Cannot read miniperl.map';
    /^\s*[\da-f:]+\s+(\w+)/i and $mapped{$1}++ foreach <MAP>;
    close MAP or die 'Cannot close miniperl.map';

    @missing = grep { !exists $mapped{$_} }
		    keys %export;
    @missing = grep { !exists $exportperlmalloc{$_} } @missing;
    delete $export{$_} foreach @missing;
}

###############################################################################

# Now all symbols should be defined because next we are going to output them.

# Start with platform specific headers:

if ($ARGS{PLATFORM} =~ /^win(?:32|ce)$/) {
    my $dll = $define{PERL_DLL} ? $define{PERL_DLL} =~ s/\.dll$//ir
	: "perl$Config{api_revision}$Config{api_version}";
    print "LIBRARY $dll\n";
    # The DESCRIPTION module definition file statement is not supported
    # by VC7 onwards.
    if ($ARGS{CCTYPE} =~ /^(?:MSVC60|GCC)$/) {
	print "DESCRIPTION 'Perl interpreter'\n";
    }
    print "EXPORTS\n";
}
elsif ($ARGS{PLATFORM} eq 'os2') {
    (my $v = $]) =~ s/(\d\.\d\d\d)(\d\d)$/$1_$2/;
    $v .= '-thread' if $Config{archname} =~ /-thread/;
    (my $dll = $define{PERL_DLL}) =~ s/\.dll$//i;
    $v .= "\@$Config{perl_patchlevel}" if $Config{perl_patchlevel};
    my $d = "DESCRIPTION '\@#perl5-porters\@perl.org:$v#\@ Perl interpreter, configured as $Config{config_args}'";
    $d = substr($d, 0, 249) . "...'" if length $d > 253;
    print <<"---EOP---";
LIBRARY '$dll' INITINSTANCE TERMINSTANCE
$d
STACKSIZE 32768
CODE LOADONCALL
DATA LOADONCALL NONSHARED MULTIPLE
EXPORTS
---EOP---
}
elsif ($ARGS{PLATFORM} eq 'aix') {
    my $OSVER = `uname -v`;
    chop $OSVER;
    my $OSREL = `uname -r`;
    chop $OSREL;
    if ($OSVER > 4 || ($OSVER == 4 && $OSREL >= 3)) {
	print "#! ..\n";
    } else {
	print "#!\n";
    }
}
elsif ($ARGS{PLATFORM} eq 'netware') {
	if ($ARGS{FILETYPE} eq 'def') {
	print "LIBRARY perl$Config{api_revision}$Config{api_version}\n";
	print "DESCRIPTION 'Perl interpreter for NetWare'\n";
	print "EXPORTS\n";
	}
}

# Then the symbols

my @symbols = $fold ? sort {lc $a cmp lc $b} keys %export : sort keys %export;
foreach my $symbol (@symbols) {
    if (PLATFORM eq 'win32' || PLATFORM eq 'wince') {
	# Remembering the origin file of each symbol is an alternative to PL_ matching
	if (substr($symbol, 0, 3) eq 'PL_') {
	    print "\t$symbol DATA\n";
	}
	else {
	    print "\t$symbol\n";
	}
    }
    elsif (PLATFORM eq 'os2') {
	printf qq(    %-31s \@%s\n),
	  qq("$symbol"), $ordinal{$symbol} || ++$sym_ord;
	printf qq(    %-31s \@%s\n),
	  qq("$exportperlmalloc{$symbol}" = "$symbol"),
	  $ordinal{$exportperlmalloc{$symbol}} || ++$sym_ord
	  if $exportperlmalloc and exists $exportperlmalloc{$symbol};
    }
    elsif (PLATFORM eq 'netware') {
	print "\t$symbol,\n";
    } else {
	print "$symbol\n";
    }
}

# Then platform specific footers.

if ($ARGS{PLATFORM} eq 'os2') {
    print <<EOP;
    dll_perlmain=main
    fill_extLibpath
    dir_subst
    Perl_OS2_handler_install

; LAST_ORDINAL=$sym_ord
EOP
}

1;


text:
import {Page} from 'ionic/ionic';
import {DataService} from '../service/data';

@Page({
  templateUrl: 'app/session-filter/session-filter.html'
})
export class SessionFilter {
  constructor(dataService: DataService) {
    this.dataService = dataService;
    this.categories = this.dataService.getCategories();
  }

  onInit() {
    // On Init reset the filters back to the last thing the user had
    this.categories.forEach((category) => {
      category.filterToApply = category.showFilter;
    });
  }

  resetFilters() {
    this.categories.forEach((category) => {
      category.filterToApply = true;
    });
  }

  applyFilters() {
    this.categories.forEach((category) => {
      category.showFilter = category.filterToApply;
    });

    this.dataService.updateCategories(this.categories);
    this.close();
  }
}


text:
package fr.isep.simizer.laws;

import org.junit.Before;

/**
 * Tests for the ZipfLaw class.
 *
 * @author Max Radermacher
 */
public class ZipfLawTest extends LawTest {
  @Before
  public void setUp() {
    law = new ZipfLaw(20, 0.25);
  }
}


text:
<input type="text" name="channel-name" placeholder="#foobar">
<input type="password" name="channel-pass" placeholder="********">

    <button type="button" class="btn btn-primary" onclick="joinChannel();toggleDialog()"></button>
    <button type="button" class="btn" onclick="toggleDialog()"></button>


text:
The development of Jitsi for Android has been stopped in early 2015. Issues and Pull Requests will not be addressed.

******

# Jitsi for Android

Jitsi for Android is an Android port of the [Jitsi] project: The most feature-rich communicator with support for encrypted audio/video, chat and presence over SIP and XMPP.

## Usage with IntelliJ

1. Make sure that you have [Java] and [Android SDK] installed on your system and [IntelliJ] version is up to date(13.0.2).
2. (Optional) Assuming that [Jitsi for desktop] project is in the same parent directory you can call "copy-jitsi-bundles" ant target. This will sync Jitsi bundles. Ant targets can be found in "Ant Build" tools window(View->Tool Windows->Ant Build).
3. Before building for the first time call "setup-libs" ant target. This will process jitsi bundles and place all required libraries in /libs folder.
4. Now you can use IntelliJ IDE to run/debug/test Jitsi for Android like any other application.

## Usage with ANT

After updating library bundles, when building for the first time or after clean:

    ant setup-libs
To make the project:

    ant make

To rebuild (clean and make):

    ant rebuild

To run the project (will install the apk and will run it on default test device):

    ant run

To make and run the project after modification:

    ant make run

## Sources

To obtain sources for .jar files located in lib folder checkout jitsi_android
 branch of jitsi and libjitsi projects.

 https://github.com/jitsi/jitsi/tree/jitsi_android

 https://github.com/jitsi/libjitsi/tree/jitsi_android

## Contribution

Before making any pull requests please see: https://jitsi.org/Documentation/FAQ#patch

[Jitsi]: https://jitsi.org/
[Java]: http://www.oracle.com/technetwork/java/javase/downloads/jdk7-downloads-1880260.html
[Android SDK]: http://developer.android.com/sdk/index.html
[IntelliJ]: http://www.jetbrains.com/idea/download/
[Jitsi for desktop]: https://github.com/jitsi/jitsi


text:
16.18: Typeahead Show results on blur example
================================
**Primary Actor**: User

**Scope**: Ngx-bootstrap DEMO / BS version 3&4

**Goal**: Show user how to use functionality to keep dropdown active on no focus

Main success scenario:
----------------------
1. User opens Typeahead demo page
2. User clicks on Show results on blur sub-menu
3. User sees typeahead input and typeahead card with "typeaheadHideResultsOnBlur: false Model:" text. User sees button with "Toggle typeaheadHideResultsOnBlur" text
4. When user clicks outside the input then drop-down stays opened
5. User clicks on the button and sees typeahead card with "typeaheadHideResultsOnBlur: false Model:" text was changed to TRUE
6. If user clicks anywhere outside the input than the drop-down closes
7. When user clicks on any item in typeahead drop-down, then typeahead container auto-fills with a selected State

Variations:
-----------
2*. User scrolls to Show results on blur sub-menu


text:
namespace caf::detail {

struct CAF_NET_EXPORT rfc6455 {
  // -- member types -----------------------------------------------------------

  struct header {
    bool fin;
    uint8_t opcode;
    uint32_t mask_key;
    uint64_t payload_len;
  };

  // -- constants --------------------------------------------------------------

  static constexpr uint8_t continuation_frame = 0x00;

  static constexpr uint8_t text_frame = 0x01;

  static constexpr uint8_t binary_frame = 0x02;

  static constexpr uint8_t connection_close = 0x08;

  static constexpr uint8_t ping = 0x09;

  static constexpr uint8_t pong = 0x0A;

  // -- utility functions ------------------------------------------------------

  static void mask_data(uint32_t key, span<char> data);

  static void mask_data(uint32_t key, byte_span data);

  static void assemble_frame(uint32_t mask_key, span<const char> data,
                             byte_buffer& out);

  static void assemble_frame(uint32_t mask_key, const_byte_span data,
                             byte_buffer& out);

  static void assemble_frame(uint8_t opcode, uint32_t mask_key,
                             const_byte_span data, byte_buffer& out);

  static ptrdiff_t decode_header(const_byte_span data, header& hdr);
};

} // namespace caf::detail


text:
function Invoke-EXRCreateAppTokenCertificate {
    [CmdletBinding()]
    param (
        [Parameter(Position = 0, Mandatory = $true)]
        [string]
        $CertName,

        [Parameter(Position = 1, Mandatory = $true)]
        [string]
        $CertFileName,

        [Parameter(Position = 2, Mandatory = $true)]
        [string]
        $ObjectId
    )
    Begin {
        if ($AccessToken -eq $null) {
            $AccessToken = Get-ProfiledToken -MailboxName $MailboxName
            if ($AccessToken -eq $null) {
                $AccessToken = Get-EXRAccessToken -MailboxName $MailboxName
            }
        }
        if ([String]::IsNullOrEmpty($MailboxName)) {
            $MailboxName = $AccessToken.mailbox
        }
        $HttpClient = Get-HTTPClient -MailboxName $MailboxName

        $Cert = New-SelfSignedCertificate -certstorelocation cert:\currentuser\my -dnsname $CertName -Provider 'Microsoft Enhanced RSA and AES Cryptographic Provider'
        $SecurePassword = Read-Host -Prompt "Enter password for Certificate File" -AsSecureString
        $CertPath = "cert:\currentuser\my\" + $Cert.Thumbprint.ToString()
        Export-PfxCertificate -cert $CertPath -FilePath $CertFileName -Password $SecurePassword
        $bin = $cert.RawData
        $base64Value = [System.Convert]::ToBase64String($bin)
        $bin = $cert.GetCertHash()
        $base64Thumbprint = [System.Convert]::ToBase64String($bin)
        $keyid = [System.Guid]::NewGuid().ToString()
        Remove-Item $CertPath
        $RequestURL = "https://graph.microsoft.com/beta/applications('" + $ObjectId + "')"
        $PostContent = @{}
        $PostContent.Add("keyCredentials", @(@{ customKeyIdentifier = $base64Thumbprint; keyId = $keyid; type = "AsymmetricX509Cert"; usage = "Verify"; key = $base64Value }))
        $JsonPost = ConvertTo-Json -Depth 10 -InputObject $PostContent
		$JSONOutput = Invoke-RestPatch -RequestURL $RequestURL -HttpClient $HttpClient -AccessToken $AccessToken -MailboxName $MailboxName -Content $JsonPost
		if($JSONOutput.IsSuccessStatusCode){
			Return "Successfully created"
		}else{
			Return $JSONOutput
		}

    }

}


text:
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
  <component name="GradleSettings">
    <option name="linkedExternalProjectsSettings">
      <GradleProjectSettings>
        <option name="distributionType" value="LOCAL" />
        <option name="externalProjectPath" value="$PROJECT_DIR$" />
        <option name="gradleHome" value="C:\Program Files\Android\Android Studio\gradle\gradle-2.10" />
        <option name="modules">
          <set>
            <option value="$PROJECT_DIR$" />
            <option value="$PROJECT_DIR$/activeandroid" />
            <option value="$PROJECT_DIR$/app" />
            <option value="$PROJECT_DIR$/framework" />
            <option value="$PROJECT_DIR$/greendao" />
            <option value="$PROJECT_DIR$/ormlite" />
            <option value="$PROJECT_DIR$/sugarorm" />
          </set>
        </option>
        <option name="myModules">
          <set>
            <option value="$PROJECT_DIR$" />
            <option value="$PROJECT_DIR$/activeandroid" />
            <option value="$PROJECT_DIR$/app" />
            <option value="$PROJECT_DIR$/framework" />
            <option value="$PROJECT_DIR$/greendao" />
            <option value="$PROJECT_DIR$/ormlite" />
            <option value="$PROJECT_DIR$/sugarorm" />
          </set>
        </option>
      </GradleProjectSettings>
    </option>
  </component>
</project>


text:
package com.senla.rakickaya.courseplanner.server;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;

import org.apache.log4j.Logger;

import com.senla.rakickaya.courseplanner.dependency.ServiceDI;
import com.senla.rakickaya.courseplanner.server.api.IFacadeWrapper;

public class DaemonClientExchanger extends Thread {
	private static final Logger logger = Logger.getLogger(DaemonClientExchanger.class.getName());

	private Socket socket;
	private BufferedReader reader;
	private BufferedWriter writer;

	private IFacadeWrapper wrapper;

	public DaemonClientExchanger(Socket socket) {
		super();
		setDaemon(true);
		this.socket = socket;
		init();

	}

	private void init() {
		try {
			reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
			writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
			wrapper = (IFacadeWrapper) ServiceDI.getInstance().getObject(IFacadeWrapper.class);
		} catch (IOException e) {
			logger.error(e.getMessage());
		}
	}

	@Override
	public void run() {
		try {
			while (true) {

				String request = reader.readLine();
				if (request == null && !socket.isConnected()) {
					break;
				}
				String response = wrapper.getResponse(request);
				writer.write(response);
				writer.newLine();
				writer.flush();
			}
		} catch (IOException e) {
			logger.info(e.getMessage());
		}

	}
}


text:
package org.springframework.cloud.service.relational;

import javax.sql.DataSource;

import org.mockito.Mock;
import org.springframework.cloud.service.AbstractCloudServiceConnectorFactoryTest;
import org.springframework.cloud.service.ServiceConnectorConfig;
import org.springframework.cloud.service.common.RelationalServiceInfo;

/**
 *
 * @author Ramnivas Laddad
 *
 */
public abstract class AbstractDataSourceFactoryTest<SI extends RelationalServiceInfo> extends AbstractCloudServiceConnectorFactoryTest<CloudDataSourceFactory, DataSource, SI> {
	@Mock DataSource mockDataSource;

	public CloudDataSourceFactory createTestCloudServiceConnectorFactory(String id, ServiceConnectorConfig config) {
		return new CloudDataSourceFactory(id, config);
	}

	public Class<DataSource> getConnectorType() {
		return DataSource.class;
	}

	public DataSource getMockConnector() {
		return mockDataSource;
	}
}


text:
from hdb import client
import argparse
import os
import sys
import time

STATUS_CHECK_INTERVAL = 0.05
MAX_CHECK_COUNT = 40

def main():
    parser = argparse.ArgumentParser(description='HDB stop script')
    parser.add_argument('port', action='store', type=int)

    args = parser.parse_args()
    server = ('localhost', args.port)
    bucket_map = client.get_bucket_map(server)

    while True:
        line = sys.stdin.readline()
        if not line:
            break

        query = parse_query(line)
        if len(query) == 0:
            print "empty query"
            continue
        if query[0][0] != '=':
            print "query should start with a specific node"
            continue

        query_server = client.locate_node(query[0][1], bucket_map)
        query_result = execute_query(query, query_server)

        print line.strip()
        if query_result["code"] == "OK":
            print query_result["result"]
        else:
            print query_result["code"]

def parse_query(line):
    result = []
    for token in line.split():
        if token[0] in ('=', '<', '>', '%'):
            result.append((token[0], int(token[1:])))
        elif token == '*':
            result.append(('*', 0))
        else:
            # simply ignore invalid tokens for now
            pass
    return result

def execute_query(query, server):
    execute_request = {"code": "QueryExecute", "query": query}
    response = client.request(server, execute_request)
    if response["code"] != "OK":
        return response

    task_key = response["task_key"]

    # check status of query until it finishes or fails
    done = False
    failed = False
    timed_out = False
    check_count = 0
    while not done and not failed and not timed_out:
        status_request = {"code": "TaskStatus", "task_key": task_key}
        response = client.request(server, status_request)
        check_count += 1
        if response["code"] != "OK" or response["status"] == "Failed":
            failed = True
        elif response["status"] == "Done":
            done = True
        elif check_count == MAX_CHECK_COUNT:
            timed_out = True
        else:
            time.sleep(STATUS_CHECK_INTERVAL)

    if failed:
        return response

    if timed_out:
        return {"code": "TimedOut"}

    # if not failed, the query succeeded and we can fetch the result
    result_request = {"code": "TaskResult", "task_key": task_key}
    response = client.request(server, result_request)
    return response

if __name__ == "__main__":
    main()


text:
require 'spec_helper'

describe SlackService do
  describe "Associations" do
    it { is_expected.to belong_to :project }
    it { is_expected.to have_one :service_hook }
  end

  describe "Validations" do
    context "active" do
      before do
        subject.active = true
      end

      it { is_expected.to validate_presence_of :webhook }
    end
  end

  describe "Execute" do
    let(:slack)   { SlackService.new }
    let(:user)    { create(:user) }
    let(:project) { create(:project) }
    let(:sample_data) { Gitlab::PushDataBuilder.build_sample(project, user) }
    let(:webhook_url) { 'https://hooks.slack.com/services/SVRWFV0VVAR97N/B02R25XN3/ZBqu7xMupaEEICInN685' }

    before do
      slack.stub(
        project: project,
        project_id: project.id,
        service_hook: true,
        webhook: webhook_url
      )

      WebMock.stub_request(:post, webhook_url)
    end

    it "should call Slack API" do
      slack.execute(sample_data)

      expect(WebMock).to have_requested(:post, webhook_url).once
    end
  end
end


text:
ActiveRecord::Schema.define do
  self.verbose = false
  create_table :posts, :force => true do |t|
    t.string :text
    t.string :status
    t.timestamps
  end
end


text:
<?php
// $Id: unit_tests.php 1661 2008-02-26 21:04:31Z pp11 $
require_once(dirname(__FILE__) . '/../autorun.php');
require_once(dirname(__FILE__) . '/../unit_tester.php');
require_once(dirname(__FILE__) . '/../shell_tester.php');
require_once(dirname(__FILE__) . '/../mock_objects.php');
require_once(dirname(__FILE__) . '/../web_tester.php');
require_once(dirname(__FILE__) . '/../extensions/pear_test_case.php');
require_once(dirname(__FILE__) . '/../extensions/phpunit_test_case.php');

class UnitTests extends TestSuite {
    function __construct() {
        parent::__construct('Unit tests');
        $path = dirname(__FILE__);
        $this->addFile($path . '/errors_test.php');
        if (version_compare(phpversion(), '5') >= 0) {
            $this->addFile($path . '/exceptions_test.php');
        }
        $this->addFile($path . '/autorun_test.php');
        $this->addFile($path . '/compatibility_test.php');
        $this->addFile($path . '/simpletest_test.php');
        $this->addFile($path . '/dumper_test.php');
        $this->addFile($path . '/expectation_test.php');
        $this->addFile($path . '/unit_tester_test.php');
        if (version_compare(phpversion(), '5', '>=')) {
            $this->addFile($path . '/reflection_php5_test.php');
        } else {
            $this->addFile($path . '/reflection_php4_test.php');
        }
        $this->addFile($path . '/mock_objects_test.php');
        if (version_compare(phpversion(), '5', '>=')) {
            $this->addFile($path . '/interfaces_test.php');
        }
        $this->addFile($path . '/collector_test.php');
        $this->addFile($path . '/adapter_test.php');
        $this->addFile($path . '/socket_test.php');
        $this->addFile($path . '/encoding_test.php');
        $this->addFile($path . '/url_test.php');
        $this->addFile($path . '/cookies_test.php');
        $this->addFile($path . '/http_test.php');
        $this->addFile($path . '/authentication_test.php');
        $this->addFile($path . '/user_agent_test.php');
        $this->addFile($path . '/parser_test.php');
        $this->addFile($path . '/tag_test.php');
        $this->addFile($path . '/form_test.php');
        $this->addFile($path . '/page_test.php');
        $this->addFile($path . '/frames_test.php');
        $this->addFile($path . '/browser_test.php');
        $this->addFile($path . '/web_tester_test.php');
        $this->addFile($path . '/shell_tester_test.php');
        $this->addFile($path . '/xml_test.php');
        $this->addFile($path . '/../extensions/testdox/test.php');
    }
}
?>


text:
LIMIT = 25000
ludic = [1]
nums = list(range(2, LIMIT))
while len(nums) > 0:
    next = nums[0]
    ludic.append(next)
    #new_nums = []
    #for i in range(len(nums)):
    #    if i % next != 0:
    #        new_nums.append(nums[i])
    #nums = new_nums
    del nums[::next] # <-- Much better!

print('Generate and show here the first 25 ludic numbers.')
print(ludic[:25])

print('How many ludic numbers are there less than or equal to 1000?')
print(sum(1 for x in ludic if x <= 1000))

print("Show the 2000..2005'th ludic numbers.")
#print(len(ludic))
print(ludic[2000-1:2005])

print('A triplet is any three numbers x, x + 2, x + 6 where all three numbers are also ludic numbers. Show all triplets of ludic numbers < 250 (Stretch goal)')
print([(x, x+2, x+6) for x in ludic if x+6 < 250 and x+2 in ludic and x+6 in ludic])


text:
package org.apache.jena.shacl.parser;

import static org.apache.commons.lang3.ObjectUtils.firstNonNull;
import static org.apache.jena.shacl.lib.ShLib.displayStr;

import java.util.*;

import org.apache.jena.atlas.logging.Log;
import org.apache.jena.ext.com.google.common.collect.ArrayListMultimap;
import org.apache.jena.ext.com.google.common.collect.Multimap;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Node;
import org.apache.jena.riot.other.G;
import org.apache.jena.shacl.engine.Parameter;
import org.apache.jena.shacl.engine.constraint.ConstraintComponentSPARQL;
import org.apache.jena.shacl.engine.constraint.SparqlComponent;
import org.apache.jena.shacl.lib.ShLib;
import org.apache.jena.shacl.vocabulary.SHACL;

public class ConstraintComponents {
    // 6.2.1 Parameter Declarations (sh:parameter)
    // 6.2.2 Label Templates (sh:labelTemplate)

    // 6.1 An Example SPARQL-based Constraint Component
    // Constraint components provide instructions to validation engines on how to
    // identify and validate constraints within a shape.
    //
    // In general, if a shape S has
    // a value for a property p, and there is a constraint component C that specifies
    // p as a parameter, and S has values for all mandatory parameters of C, then the
    // set of these parameter values (including the optional parameters) declare a
    // constraint and the validation engine uses a suitable validator from C to
    // perform the validation of this constraint.
    //
    // In the example above,
    // sh:PatternConstraintComponent declares the mandatory parameter sh:pattern, the
    // optional parameter sh:flags, and a validator that can be used to perform
    // validation against either node shapes or property shapes.

    // 6.3 Validation with SPARQL-based Constraint Components

    /*package*/ Multimap<Node, SparqlComponent> paramPathToComponents = ArrayListMultimap.create();
    /*package*/ Set<Parameter> parameters = new HashSet<>();

    /*package*/ ConstraintComponents() {}
    /*package*/ boolean hasParameters() { return parameters.isEmpty(); }

    /** Stage 1 : find all Constraint Components */
    public static ConstraintComponents parseSparqlConstraintComponents(Graph shapesGraph) {
        ConstraintComponents x = new ConstraintComponents();
        // SHACL.SPARQLConstraintComponent is not a subclass of SHACL.ConstraintComponent
        // SHACL.SPARQLConstraintComponent is an instance of SHACL.ConstraintComponent.
        // Need to process by actual property present.
        G.allNodesOfTypeRDFS(shapesGraph, SHACL.ConstraintComponent).forEach(sccNode->{
            SparqlComponent c = sparqlConstraintComponent(shapesGraph, sccNode);
            if ( c != null ) {
                for ( Parameter p : c.getParams() ) {
                    x.paramPathToComponents.put(p.getParameterPath(), c);
                    x.parameters.add(p);
                }
            }
        });
        return x;
    }

    /**
     * Stage 2 :
     * Build all the Constraints for Validators for this shape.
     * Must have all the required parameters (else ignore).
     */
    public static List<Constraint> processShape(Graph shapesGraph, ConstraintComponents components, Shape shape) {
        List<Constraint> constraints = new ArrayList<>();

        // Only add set of requires once, not once per parameter.
        Set<Set<Node>> seen = new HashSet<>();

        for ( Parameter param : components.parameters ) {
            if ( param.isOptional() )
                continue;
            // The parameter's property node
            Node paramPath = param.getParameterPath();

            // Does the shape use the parameter?
            boolean b = G.contains(shapesGraph, shape.getShapeNode(), paramPath, null);
            if ( !b )
                continue;

//          Alternative approach
//            // Find all shapes uses the parameter.
//            List<Node> shapes = G1.find(shapesGraph, null, paramPath, null).mapWith(Triple::getSubject).toList();
            Node shNode = shape.getShapeNode();

            // All components with this parameter.
            Collection<SparqlComponent> sccs = components.paramPathToComponents.get(paramPath);
            // Which shapes conform (have all required parameters)?
            // And we have not seen before for this shape?
            sccs.forEach(scc->{
                List<Node> required = scc.getRequiredParameters();

                // Check not seen.
                Set<Node> x = Set.copyOf(required);
                if ( seen.contains(x) ) {
                    return;
                }
                seen.add(x);

                if ( Parameters.doesShapeHaveAllParameters(shapesGraph, shape.getShapeNode(), required) ) {
                    // shape -> parameters
                    // Is this a cross product and can it be avoided?
                    Multimap<Parameter, Node> parameterValues = constraintParameterValues(shapesGraph, shNode, scc);
                    // [PARSE]
                    // Syntax rule: https://www.w3.org/TR/shacl/#syntax-rule-multiple-parameters
                    //  If there are >1 parameters, each must be single valued.

                    if ( parameterValues.keySet().size() > 1 ) {
                        Map<Parameter, Node> parameterMap = new HashMap<>();
                        parameterValues.asMap().forEach((p,values)->{
                           if ( values.size() > 1 )
                               throw new ShaclParseException("Multiple values for parameter "+p+" in constraint with multiple parameters");
                        });
                    }
                    ConstraintComponentSPARQL constraintComponentSPARQL = new ConstraintComponentSPARQL(scc, parameterValues);
                    constraints.add(constraintComponentSPARQL);
                }
                else {
                    // Incomplete.
                    //System.out.println("shape: no: "+shNode);
                }
            });
        }
        return constraints;
    }

    private static Multimap<Parameter, Node> constraintParameterValues(Graph shapesGraph, Node shNode, SparqlComponent scc) {
        Node nodeValidatorNode = G.getZeroOrOneSP(shapesGraph, shNode, SHACL.nodeValidator);
        Node propertyValidatorNode = G.getZeroOrOneSP(shapesGraph, shNode, SHACL.propertyValidator);
        Node validatorNode = G.getZeroOrOneSP(shapesGraph, shNode, SHACL.validator);
        Node vNode = firstNonNull(nodeValidatorNode, propertyValidatorNode, validatorNode);
        Multimap<Parameter, Node> parameterValues = Parameters.parameterValues(shapesGraph, shNode, scc);
        return parameterValues;
    }

    /** This handles all ConstraintComponents; only SPARQL ones are currently supported. */
    private static SparqlComponent sparqlConstraintComponent(Graph shapesGraph, Node constraintComponentNode) {
        if ( SHACL.JSConstraintComponent.equals(constraintComponentNode) )
            return null;
        List<Parameter> params = Parameters.parseParameters(shapesGraph, constraintComponentNode);

        // 6.2.3 Validators
        //
        // For every supported shape type (i.e., property shape or node shape) the constraint component
        // declares a suitable validator. For a given constraint, a validator is selected from the
        // constraint component using the following rules, in order:
        //
        //     1. For node shapes, use one of the values of sh:nodeValidator, if present.
        //     2. For property shapes, use one of the values of sh:propertyValidator, if present.
        //     3. Otherwise, use one of the values of sh:validator.
        //
        // If no suitable validator can be found, a SHACL-SPARQL processor ignores the constraint.
        //
        // SHACL-SPARQL includes two types of validators, based on SPARQL SELECT (for sh:nodeValidator
        // and sh:propertyValidator) or SPARQL ASK queries (for sh:validator).

        SparqlComponent x = possibleValidator(shapesGraph, constraintComponentNode, SHACL.nodeValidator, params);
        if ( x == null )
            x = possibleValidator(shapesGraph, constraintComponentNode, SHACL.propertyValidator, params);
        if ( x == null )
            x = possibleValidator(shapesGraph, constraintComponentNode, SHACL.validator, params);
        return x;
    }

    private static SparqlComponent possibleValidator(Graph shapesGraph, Node constraintComponentNode, Node vProperty, List<Parameter> params) {
        List<Node> x = G.listSP(shapesGraph, constraintComponentNode, vProperty);
        if ( x.isEmpty() )
            return null;
        for(Node valNode : x ) {
            SparqlComponent cs = possibleSparqlValidator(shapesGraph, valNode, params, constraintComponentNode);
            if ( cs != null )
                return cs;
        }
        return null;
    }

    private static SparqlComponent possibleSparqlValidator(Graph shapesGraph, Node valNode, List<Parameter> params,
                                                           /* for reporting */ Node constraintComponentNode) {
        // Check for SHACL-JS
        Node xJSFunctionName = G.getZeroOrOneSP(shapesGraph, valNode, SHACL.jsFunctionName);
        if ( xJSFunctionName != null )
            Log.warn(ConstraintComponents.class, "Found javascript validator - ignored (JavaScript not currently supported)");

        // One of sh:select or sh:ask.
        Node xSelect = G.getZeroOrOneSP(shapesGraph, valNode, SHACL.select);
        Node xAsk = G.getZeroOrOneSP(shapesGraph, valNode, SHACL.ask);
        if ( xSelect == null && xAsk == null )
            return null;
        if ( xSelect != null && xAsk != null )
            throw new ShaclParseException("SparqlConstraintComponent: Multiple SPARQL queries: "+displayStr(constraintComponentNode));
        String prefixes = ShLib.prefixes(shapesGraph, valNode);
        String queryString = firstNonNull(xSelect, xAsk).getLiteralLexicalForm().trim();
        String message = asString(G.getZeroOrOneSP(shapesGraph, valNode, SHACL.message));
        if ( ! prefixes.isEmpty() )
            queryString = prefixes+"\n"+queryString;
        boolean isSelect = (xSelect!=null);
        SparqlComponent cs = SparqlComponent.constraintComponent(constraintComponentNode, queryString, params, message);
        if ( cs.getQuery().isSelectType() != isSelect )
            throw new ShaclParseException("Query type does not match property");
        return cs;
    }

    private static String asString(Node x) {
        if ( x == null )
            return null;
        if ( ! x.isLiteral() )
            return null;
        return x.getLiteralLexicalForm();
    }
}