Add Unix Stream codec, Split flow from FDK, update integration tests

api/pom.xml

@@ -11,6 +11,18 @@
 
     <artifactId>api</artifactId>
 
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
     <build>
         <plugins>
             <plugin>

api/src/main/java/com/fnproject/fn/api/FnFeature.java

@@ -0,0 +1,21 @@
+package com.fnproject.fn.api;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation to be used in user function classes to enable runtime-wide feature.
+ *
+ * Runtime features are initialized at the point that the function class is loaded but prior to the call chain.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface FnFeature {
+    /**
+     * The feature class to load this must have a zero-arg public constructor
+     * @return feature class
+     */
+    Class<? extends  RuntimeFeature> value();
+}

api/src/main/java/com/fnproject/fn/api/FunctionInvoker.java

@@ -6,6 +6,22 @@
  * Handles the invocation of a given function call
  */
 public interface FunctionInvoker {
+    /**
+     * Phase determines a loose ordering for invocation handler processing
+     * this should be used with {@link RuntimeContext#addInvoker(FunctionInvoker, Phase)} to add new invoke handlers to a runtime
+     */
+    enum Phase {
+        /**
+         * The Pre-Call phase runs before the main function call, all {@link FunctionInvoker} handlers added at this phase are tried prior to calling the {@link Phase#Call} phase
+         * This phase is typically used for handlers that /may/ intercept the request based on request attributes
+         */
+        PreCall,
+        /**
+         * The Call Phase indicates invokers that should handle call values - typically a given runtime will only be handled by one of these
+         */
+        Call
+    }
+
     /**
      * Optionally handles an invocation chain for this function
      * <p>

api/src/main/java/com/fnproject/fn/api/Headers.java

@@ -1,16 +1,51 @@
 package com.fnproject.fn.api;
 
 import java.util.*;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
 
 /**
- * Represents the headers on an HTTP request or response. Multiple headers with the same key are collapsed into a single
- * entry where the values are concatenated by commas as per the HTTP spec (RFC 7230).
+ * Represents a set of String-String[] header attributes, per HTTP headers.
+ * <p>
+ * Internally header keys are always canonicalized using HTTP header conventions
+ * <p>
+ * <p>
+ * Headers are immutable
+ * <p>
+ * keys are are stored  and compared in a case-insensitive way
  */
 public final class Headers {
-    private Map<String, String> headers;
+    private static final Headers emptyHeaders = new Headers(Collections.emptyMap());
+    private Map<String, List<String>> headers;
+
+    private Headers(Map<String, List<String>> headersIn) {
+        this.headers = headersIn;
+    }
+
+    private static Pattern headerName = Pattern.compile("[A-Za-z0-9!#%&'*+-.^_`|~]+");
+
+
+    /**
+     * Calculates the canonical key  (cf RFC 7230) for a header
+     * <p>
+     * If the header contains invalid characters it returns the original header
+     *
+     * @param key the header key to canonicalise
+     * @return a canonical key or the original key if the input contains invalid character
+     */
+    public static String canonicalKey(String key) {
+        if (!headerName.matcher(key).matches()) {
+            return key;
+        }
+        String parts[] = key.split("-", -1);
+        for (int i = 0; i < parts.length; i++) {
+            String p = parts[i];
+            if (p.length() > 0) {
+                parts[i] = p.substring(0, 1).toUpperCase() + p.substring(1).toLowerCase();
+            }
+        }
+        return String.join("-", parts);
 
-    private Headers(Map<String, String> headers) {
-        this.headers = headers;
     }
 
     /**
@@ -21,6 +56,23 @@ private Headers(Map<String, String> headers) {
      * @return {@code Headers} built from headers map
      */
     public static Headers fromMap(Map<String, String> headers) {
+        Objects.requireNonNull(headers, "headersIn");
+        Map<String, List<String>> h = new HashMap<>();
+        headers.forEach((k, v) -> h.put(canonicalKey(k), Collections.singletonList(v)));
+        return new Headers(Collections.unmodifiableMap(new HashMap<>(h)));
+    }
+
+    /**
+     * Build a headers object from a map composed of (name, value) entries, we take a copy of the map and
+     * disallow any further modification
+     *
+     * @param headers underlying collection of header entries to copy
+     * @return {@code Headers} built from headers map
+     */
+    public static Headers fromMultiHeaderMap(Map<String, List<String>> headers) {
+        Map<String, List<String>> hm = new HashMap<>();
+
+        headers.forEach((k, vs) -> hm.put(canonicalKey(k), new ArrayList<>(vs)));
         return new Headers(Collections.unmodifiableMap(new HashMap<>(Objects.requireNonNull(headers))));
     }
 
@@ -30,64 +82,177 @@ public static Headers fromMap(Map<String, String> headers) {
      * @return empty headers
      */
     public static Headers emptyHeaders() {
-        return new Headers(Collections.emptyMap());
+        return emptyHeaders;
     }
 
 
     /**
-     * Creates a new headers object with the specified header added
+     * Creates a new headers object with the specified header added - if a  header with the same key existed it the new value is appended
+     * <p>
+     * This will overwrite an existing header with an exact name match
      *
+     * @param key new header key
+     * @param v1  new header value
+     * @param vs  additional header values to set
+     * @return a new headers object with the specified header added
+     */
+    public Headers addHeader(String key, String v1, String... vs) {
+        Objects.requireNonNull(key, "key");
+        Objects.requireNonNull(key, "value");
+
+        String canonKey = canonicalKey(key);
+
+        Map<String, List<String>> nm = new HashMap<>(headers);
+        List<String> current = nm.get(canonKey);
+
+        if (current == null) {
+            List<String> s = new ArrayList<>();
+            s.add(v1);
+            s.addAll(Arrays.asList(vs));
+
+            nm.put(canonKey, Collections.unmodifiableList(s));
+        } else {
+            List<String> s = new ArrayList<>(current);
+            s.add(v1);
+            s.addAll(Arrays.asList(vs));
+            nm.put(canonKey, Collections.unmodifiableList(s));
+        }
+        return new Headers(nm);
+
+    }
+
+    /**
+     * Creates a new headers object with the specified header set - this overwrites any existin values
+     * <p>
      * This will overwrite an existing header with an exact name match
+     *
      * @param key new header key
-     * @param value new header value
+     * @param v1  new header value
+     * @param vs  more header values to set
      * @return a new headers object with the specified header added
      */
-    public Headers withHeader(String key, String value){
-        Map<String,String> newHeaders = new HashMap<>();
-        newHeaders.putAll(getAll());
-        newHeaders.put(key,value);
-        return new Headers(newHeaders);
+    public Headers setHeader(String key, String v1, String... vs) {
+        Objects.requireNonNull(key, "key");
+        Objects.requireNonNull(v1, "v1");
+        Stream.of(vs).forEach((v) -> Objects.requireNonNull(v, "vs"));
+
+        Map<String, List<String>> nm = new HashMap<>(headers);
+        List<String> s = new ArrayList<>();
+        s.add(v1);
+        s.addAll(Arrays.asList(vs));
+        nm.put(canonicalKey(key), Collections.unmodifiableList(s));
+        return new Headers(Collections.unmodifiableMap(nm));
+    }
+
+
+    /**
+     * Creates a new headers object with the specified headers set - this overwrites any existin values
+     * <p>
+     * This will overwrite an existing header with an exact name match
+     *
+     * @param key new header key
+     * @param vs  header values to set
+     * @return a new headers object with the specified header added
+     */
+    public Headers setHeader(String key, Collection<String> vs) {
+        Objects.requireNonNull(key, "key");
+        Objects.requireNonNull(vs, "vs");
+        if (vs.size() == 0) {
+            throw new IllegalArgumentException("can't set keys to an empty list");
+        }
+        vs.forEach((v) -> Objects.requireNonNull(v, "vs"));
+
+        Map<String, List<String>> nm = new HashMap<>(headers);
+        nm.put(canonicalKey(key), Collections.unmodifiableList(new ArrayList<>(vs)));
+        return new Headers(Collections.unmodifiableMap(nm));
+
+    }
+
+    /**
+     * Creates a new headers object with the specified headers remove - this overwrites any existin values
+     * <p>
+     * This will overwrite an existing header with an exact name match
+     *
+     * @param key new header key
+     * @return a new headers object with the specified header removed
+     */
+    public Headers removeHeader(String key) {
+        Objects.requireNonNull(key, "key");
+
+        String canonKey = canonicalKey(key);
+        if (!headers.containsKey(canonKey)) {
+            return this;
+        }
+
+        Map<String, List<String>> nm = new HashMap<>(headers);
+        nm.remove(canonKey);
+        return new Headers(Collections.unmodifiableMap(nm));
+
     }
 
     /**
      * Returns the header matching the specified key. This matches headers in a case-insensitive way and substitutes
      * underscore and hyphen characters such that : "CONTENT_TYPE" and "Content-type" are equivalent. If no matching
      * header is found then {@code Optional.empty} is returned.
      * <p>
-     * Multiple headers are collapsed by {@code fn} into a single header entry delimited by commas (see
-     * <a href="https://tools.ietf.org/html/rfc7230#section-3.2.2">RFC7230 Sec 3.2.2</a> for details), for example
-     *
-     * <pre>
-     *     Accept: text/html
-     *     Accept: text/plain
-     * </pre>
-     *
-     * is collapsed into
-     *
-     * <pre>
-     *     Accept: text/html, text/plain
-     * </pre>
+     * When multiple headers are present then the first value is returned- see { #getAllValues(String key)} to get all values for a header
      *
      * @param key match key
      * @return a header matching key or empty if no header matches.
      * @throws NullPointerException if {@code key} is null.
      */
     public Optional<String> get(String key) {
         Objects.requireNonNull(key, "Key cannot be null");
-        return getAll().entrySet().stream()
-                .filter((e) -> e.getKey()
-                        .replaceAll("-", "_")
-                        .equalsIgnoreCase(key.replaceAll("-", "_")))
-                .map(Map.Entry::getValue)
-                .findFirst();
+        String canonKey = canonicalKey(key);
+
+        return headers.entrySet().stream()
+          .filter((e) -> e.getKey()
+            .equals(canonKey))
+          .map(Map.Entry::getValue)
+          .map((v) -> v.get(0))
+          .findFirst();
+    }
+
+    /**
+     * Returns a collection  of current header keys
+     *
+     * @return a collection of keys
+     */
+    public Collection<String> keys() {
+        return headers.keySet();
     }
 
     /**
-     * The function invocation headers passed on the request
+     * Returns the headers as a  map
      *
-     * @return a map of Invocation headers.
+     * @return a map of key-values
      */
-    public Map<String, String> getAll() {
+    public Map<String, List<String>> asMap() {
         return headers;
     }
+
+    public List<String> getAllValues(String key) {
+        return headers.getOrDefault(key, Collections.emptyList());
+    }
+
+    public int hashCode() {
+        return headers.hashCode();
+    }
+
+
+    public boolean equals(Object other) {
+        if (!(other instanceof Headers)) {
+            return false;
+        }
+        if (other == this) {
+            return true;
+        }
+        return headers.equals(((Headers) other).headers);
+    }
+
+    @Override
+    public String toString() {
+        return Objects.toString(headers);
+    }
+
 }