diff --git a/.github/workflows/gradle.yml b/.github/workflows/gradle.yml
index e515180..84b3afd 100644
--- a/.github/workflows/gradle.yml
+++ b/.github/workflows/gradle.yml
@@ -7,7 +7,7 @@ on:
   push:
     branches: [ main ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, release/develop ]
 
 jobs:
   build:
@@ -23,4 +23,4 @@ jobs:
     - name: Grant execute permission for gradlew
       run: chmod +x gradlew
     - name: Build with Gradle
-      run: ./gradlew --console=plain build
+      run: ./gradlew -x shadowJar --console=plain build
diff --git a/.run/Doer kfeed - simple ingest.run.xml b/.run/Doer pipeline grpc service.run.xml
similarity index 58%
rename from .run/Doer kfeed - simple ingest.run.xml
rename to .run/Doer pipeline grpc service.run.xml
index 6c234f8..b4761c4 100644
--- a/.run/Doer kfeed - simple ingest.run.xml	
+++ b/.run/Doer pipeline grpc service.run.xml	
@@ -1,11 +1,11 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="Doer kfeed - simple ingest" type="Application" factoryName="Application">
+  <configuration default="false" name="Doer pipeline grpc service" type="Application" factoryName="Application">
     <option name="MAIN_CLASS_NAME" value="io.github.s7i.doer.Doer" />
     <module name="doer.main" />
-    <option name="PROGRAM_PARAMETERS" value="kfeed -y src/test/resources/simple-ingest-dry-run.yml" />
+    <option name="PROGRAM_PARAMETERS" value="pipeline" />
     <extension name="coverage">
       <pattern>
-        <option name="PATTERN" value="io.github.s7i.doer.*" />
+        <option name="PATTERN" value="io.github.s7i.doer.command.*" />
         <option name="ENABLED" value="true" />
       </pattern>
     </extension>
diff --git a/.run/Doer pipeline ingest.run.xml b/.run/Doer pipeline ingest.run.xml
new file mode 100644
index 0000000..45b26f0
--- /dev/null
+++ b/.run/Doer pipeline ingest.run.xml	
@@ -0,0 +1,16 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="Doer pipeline ingest" type="Application" factoryName="Application">
+    <option name="MAIN_CLASS_NAME" value="io.github.s7i.doer.Doer" />
+    <module name="doer.main" />
+    <option name="PROGRAM_PARAMETERS" value="./docs/pipeline/grpc-backend/ingest.yml" />
+    <extension name="coverage">
+      <pattern>
+        <option name="PATTERN" value="io.github.s7i.doer.command.*" />
+        <option name="ENABLED" value="true" />
+      </pattern>
+    </extension>
+    <method v="2">
+      <option name="Make" enabled="true" />
+    </method>
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/.run/Doer pipeline sink.run.xml b/.run/Doer pipeline sink.run.xml
new file mode 100644
index 0000000..4750996
--- /dev/null
+++ b/.run/Doer pipeline sink.run.xml	
@@ -0,0 +1,16 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="Doer pipeline sink" type="Application" factoryName="Application">
+    <option name="MAIN_CLASS_NAME" value="io.github.s7i.doer.Doer" />
+    <module name="doer.main" />
+    <option name="PROGRAM_PARAMETERS" value="./docs/pipeline/grpc-backend/console-sink.yml" />
+    <extension name="coverage">
+      <pattern>
+        <option name="PATTERN" value="io.github.s7i.doer.command.*" />
+        <option name="ENABLED" value="true" />
+      </pattern>
+    </extension>
+    <method v="2">
+      <option name="Make" enabled="true" />
+    </method>
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/build.gradle b/build.gradle
index 90cf9a0..2aff6e3 100644
--- a/build.gradle
+++ b/build.gradle
@@ -72,6 +72,11 @@ application {
     mainClass.set('io.github.s7i.doer.Doer')
 }
 
+shadowJar {
+    // https://github.com/grpc/grpc-java/issues/5493#issuecomment-478500418
+    mergeServiceFiles()
+}
+
 tasks.withType(JavaCompile) {
     options.compilerArgs << '-Xlint:unchecked'
     options.deprecation = true
diff --git a/docs/pipeline/grpc-backend/console-sink.yml b/docs/pipeline/grpc-backend/console-sink.yml
new file mode 100644
index 0000000..edbd90d
--- /dev/null
+++ b/docs/pipeline/grpc-backend/console-sink.yml
@@ -0,0 +1,16 @@
+version: v1
+kind: sink
+params:
+  doer.pipeline.backend: grpc
+  doer.pipeline.backend.target: localhost:6565
+  doer.pipeline.sink: true
+  doer.pipeline.bind: from-ingest-to-sink
+  doer.pipeline.id: id-sink1
+  mykaf: |+
+    bootstrap.servers=localhost:9092
+    client.id=doer.sink
+spec:
+  - output: doer://console
+#    enabled: false
+  - output: kafka://mykaf/test-sink123
+#    enable: false
diff --git a/docs/pipeline/grpc-backend/ingest.yml b/docs/pipeline/grpc-backend/ingest.yml
new file mode 100644
index 0000000..56a4872
--- /dev/null
+++ b/docs/pipeline/grpc-backend/ingest.yml
@@ -0,0 +1,11 @@
+version: v1
+kind: ingest
+param:
+  doer.output: pipeline://grpc
+  doer.pipeline.backend: grpc
+  doer.pipeline.backend.target: localhost:6565
+ingest:
+  - record: record 1 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+  - record: record 2 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+  - record: record 3 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+    key: key-record 3
diff --git a/docs/pipeline/grpc-backend/readme.md b/docs/pipeline/grpc-backend/readme.md
new file mode 100644
index 0000000..8c06331
--- /dev/null
+++ b/docs/pipeline/grpc-backend/readme.md
@@ -0,0 +1,17 @@
+### Running from the console
+
+1. Sink:
+
+    ```bash
+    doer console-sink.yml
+    
+    ```
+
+2. Service:
+    ```bash
+    doer pipeline
+    ```
+3. Records ingest
+    ```bash
+    doer ingest.yml
+    ```
diff --git a/docs/pipeline/pipeline-record.sh b/docs/pipeline/pipeline-record.sh
new file mode 100755
index 0000000..64dc370
--- /dev/null
+++ b/docs/pipeline/pipeline-record.sh
@@ -0,0 +1,61 @@
+#!/bin/bash
+
+PROTO_SET=../../proto-doer/build/descriptors/main.desc
+set -e
+
+doer misc pipeline record --help
+
+print_color() {
+    local TEXT=$1
+    local CLR_NAME=$2
+    local RED="\e[31m"
+    local GREEN="\e[32m"
+    local ENDCOLOR="\e[0m"
+    local COLOR=${!CLR_NAME}
+    echo -e "${COLOR}${TEXT}${ENDCOLOR}"
+}
+
+green() {
+    print_color $1 "GREEN"
+}
+
+make_record() {
+    doer misc pipeline record \
+      --meta test.data=true \
+      --key my-key \
+      --data "some test data"
+}
+
+make_record_pl() {
+    doer misc pipeline record \
+      --meta test.data=true \
+      --key my-key \
+      --data "some test data" \
+      --pipeline-load
+}
+
+green "[HEX:record]"
+make_record | xxd
+
+# https://github.com/protocolbuffers/protoscope
+green "[protoscope]"
+make_record | protoscope
+
+green "[protoscope:print-field-names]"
+
+make_record | protoscope \
+   -descriptor-set ${PROTO_SET} \
+   -message-type io.github.s7i.doer.proto.Record \
+   -print-field-names
+
+green "[HEX:pipeline-load]"
+make_record_pl | xxd
+
+green "[protoscope:pipeline-load]"
+make_record_pl | protoscope
+
+green "[protoscope:pipeline-load:print-field-names]"
+make_record_pl | protoscope \
+   -descriptor-set ${PROTO_SET} \
+   -message-type io.github.s7i.doer.pipeline.proto.PipelineLoad \
+   -print-field-names
\ No newline at end of file
diff --git a/docs/pipeline/readme.md b/docs/pipeline/readme.md
new file mode 100644
index 0000000..dd85bb1
--- /dev/null
+++ b/docs/pipeline/readme.md
@@ -0,0 +1,39 @@
+# Pipelines
+
+Sample of pipeline specification:
+
+```yaml
+# Pipeline manifest draft
+version: v1
+kind: pipeline
+params:
+  doer.pipeline.backend: kafka
+  doer.pipeline.backend.kafka.properties: |+
+    bootstrap.servers=kafka:9093
+spec:
+  - pipeline: a | b
+  - name: a
+    description: "a" as instance of source
+    manifest-file: source.yml
+  - name: b
+    description: "b" as instance of sink
+    manifest-file: sink.yml
+---
+
+version: v2
+kind: pipeline
+params:
+  doer.pipeline.backend: kafka
+  doer.pipeline.backend.kafka.properties: |+
+    bootstrap.servers=kafka:9093
+spec:
+  pipeline:
+    - flow: a | b
+  elements:
+    - name: a
+      description: "a" as instance of source
+      manifest-file: source.yml
+    - name: b
+      description: "b" as instance of sink
+      manifest-file: sink.yml
+```
\ No newline at end of file
diff --git a/proto-doer/src/main/proto/doer.proto b/proto-doer/src/main/proto/doer.proto
index 78984f1..c381f8a 100644
--- a/proto-doer/src/main/proto/doer.proto
+++ b/proto-doer/src/main/proto/doer.proto
@@ -5,6 +5,7 @@ package io.github.s7i.doer.proto;
 option java_multiple_files = true;
 
 import "google/protobuf/wrappers.proto";
+import "google/protobuf/empty.proto";
 
 message DoerVersion {
   string name = 1;
@@ -17,3 +18,15 @@ message Record {
   google.protobuf.BytesValue data = 4;
 }
 
+service DoerService {
+
+  rpc version (google.protobuf.Empty) returns (Sone);
+  rpc exec(Sone) returns (Sone);
+  rpc pulse(google.protobuf.Empty) returns (stream Sone);
+
+}
+
+message Sone {
+  string s = 1;
+}
+
diff --git a/proto-doer/src/main/proto/pipeline.proto b/proto-doer/src/main/proto/pipeline.proto
index f58ae9c..89ef9ef 100644
--- a/proto-doer/src/main/proto/pipeline.proto
+++ b/proto-doer/src/main/proto/pipeline.proto
@@ -7,7 +7,22 @@ import "google/protobuf/any.proto";
 option java_multiple_files = true;
 
 service PipelineService {
+  rpc exchangeMeta(MetaOp) returns (MetaOp);
   rpc publish(PipelinePublishRequest) returns (PipelinePublishResponse);
+  rpc subscribe(MetaOp) returns (stream PipelineLoad);
+}
+
+message MetaOp {
+  message Request {
+    string name = 1;
+    map<string, string> parameters = 2;
+  }
+  message Response {
+    string status = 1;
+    map<string, string> parameters = 2;
+  }
+  Request request = 1;
+  Response response = 2;
 }
 
 message PipelinePublishRequest {
diff --git a/spec/manifest/pipeline.yml b/spec/manifest/pipeline.yml
deleted file mode 100644
index d4b41e5..0000000
--- a/spec/manifest/pipeline.yml
+++ /dev/null
@@ -1,13 +0,0 @@
-# Pipeline manifest draft
-version: "0.1"
-kind: pipeline
-pipeline: a | b
-spec:
-  - name: a
-    description: "a" as instance of kafka-dump, making output to doer://pipeline
-    manifest:
-      file: kdump.yml
-  - name: b
-    description: "b" as instance of kafka-feed, read from a pipeline
-    manifest:
-      file: kfeed.yml
\ No newline at end of file
diff --git a/src/main/java/io/github/s7i/doer/Context.java b/src/main/java/io/github/s7i/doer/Context.java
index fee2801..bd72ca9 100644
--- a/src/main/java/io/github/s7i/doer/Context.java
+++ b/src/main/java/io/github/s7i/doer/Context.java
@@ -1,19 +1,27 @@
 package io.github.s7i.doer;
 
-import static io.github.s7i.doer.Doer.console;
-import static java.util.Objects.requireNonNull;
-
-import io.github.s7i.doer.domain.output.*;
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.domain.output.OutputBuilder;
+import io.github.s7i.doer.domain.output.OutputFactory;
+import io.github.s7i.doer.domain.output.OutputProvider;
+import io.github.s7i.doer.pipeline.Pipeline;
 import io.github.s7i.doer.util.ParamFlagExtractor;
 import io.github.s7i.doer.util.PropertyResolver;
 import io.github.s7i.doer.util.QuitWatcher;
-import java.nio.file.Path;
-import java.util.Collections;
-import java.util.Map;
 import lombok.Builder;
 import lombok.Builder.Default;
 import lombok.Getter;
 
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.github.s7i.doer.Doer.console;
+import static java.util.Objects.nonNull;
+import static java.util.Objects.requireNonNull;
+
 public interface Context extends ParamFlagExtractor {
 
     @Builder
@@ -28,10 +36,11 @@ class InitialParameters {
     class Initializer {
 
         static {
-            Runtime.getRuntime().addShutdownHook(new Thread(Initializer::shutdown, "shutdown"));
+            Runtime.getRuntime().addShutdownHook(new Thread(Globals.INSTANCE::stopAll, "shutdown"));
         }
 
         public Initializer(InitialParameters parameters) {
+            Pipeline.initFrom(parameters::getParams);
             var scope = Globals.INSTANCE.getScope();
             scope.setRoot(parameters::getWorkDir);
             scope.setParams(parameters::getParams);
@@ -39,10 +48,8 @@ public Initializer(InitialParameters parameters) {
             new QuitWatcher().watchForQuit(() -> System.exit(Doer.EC_QUIT));
         }
 
-        private static void shutdown() {
-            console().info("Init shutdown procedure...");
-            Globals.INSTANCE.stopHooks.forEach(Runnable::run);
-            console().info("Shutdown completed.");
+        public Context context() {
+            return Globals.INSTANCE;
         }
     }
 
@@ -50,6 +57,16 @@ default void addStopHook(Runnable runnable) {
         Globals.INSTANCE.stopHooks.add(runnable);
     }
 
+    default void stopAll() {
+        console().info("Init shutdown procedure...");
+        Globals.INSTANCE.stopHooks.forEach(Runnable::run);
+        console().info("Shutdown completed.");
+    }
+
+    default void stopAllSilent() {
+        Globals.INSTANCE.stopHooks.forEach(Runnable::run);
+    }
+
     default OutputFactory getOutputFactory() {
         return Globals.INSTANCE.getScope().getOutputFactory();
     }
@@ -71,4 +88,17 @@ default Map<String, String> getParams() {
     default PropertyResolver getPropertyResolver() {
         return new PropertyResolver(getParams());
     }
+
+    default Optional<Pipeline> lookupPipeline() {
+        var pipeline = Globals.INSTANCE.pipeline;
+        if (nonNull(pipeline) && pipeline.isEnabled()) {
+            return Optional.of(pipeline);
+        }
+        return Optional.empty();
+    }
+
+    default void shareParams(Map<String, String> otherParams) {
+        var copy = new HashMap<>(otherParams);
+        Globals.INSTANCE.getScope().setParams(() -> copy);
+    }
 }
diff --git a/src/main/java/io/github/s7i/doer/Doer.java b/src/main/java/io/github/s7i/doer/Doer.java
index cc42631..59a2226 100644
--- a/src/main/java/io/github/s7i/doer/Doer.java
+++ b/src/main/java/io/github/s7i/doer/Doer.java
@@ -4,6 +4,9 @@
 import io.github.s7i.doer.command.dump.KafkaDump;
 import io.github.s7i.doer.command.util.CommandManifest;
 import io.github.s7i.doer.command.util.Misc;
+import io.github.s7i.doer.domain.ServiceEntrypoint;
+import io.github.s7i.doer.domain.grpc.GrpcServer;
+import io.github.s7i.doer.pipeline.PipelineService;
 import io.github.s7i.doer.util.Banner;
 import io.github.s7i.doer.util.GitProps;
 import org.slf4j.Logger;
@@ -11,6 +14,7 @@
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 
+import java.io.IOException;
 import java.util.Arrays;
 
 @Command(name = "doer", description = "let's do big things...", subcommands = {
@@ -21,10 +25,12 @@
       Rocks.class,
       GrpcHealth.class,
       ZooSrv.class,
+      PipelineService.class,
       Misc.class})
 public class Doer implements Runnable, Banner {
 
-    static final Logger CONSOLE = LoggerFactory.getLogger("doer.console");
+    public static final String DOER_CONSOLE = "doer.console";
+    static final Logger CONSOLE = LoggerFactory.getLogger(DOER_CONSOLE);
     public static final String FLAGS = "doer.flags";
     public static final String FLAG_USE_TRACING = "trace";
     public static final String FLAG_SEND_AND_FORGET = "send-and-forget";
@@ -43,6 +49,17 @@ public static Logger console() {
     @CommandLine.Option(names = {"-v", "--version"})
     private boolean showVersion;
 
+    @Command(name = "srv-grpc")
+    public int service(int port) {
+        try {
+            new GrpcServer(port, new ServiceEntrypoint()).startServer();
+            return 0;
+        } catch (IOException e) {
+            console().error("grpc-server", e);
+            return EC_ERROR;
+        }
+    }
+
     @Override
     public void run() {
         printBanner();
@@ -61,8 +78,10 @@ public static void main(String[] args) {
               ? new CommandManifest()
               : new Doer();
 
-        new CommandLine(command)
+        var exitCode = new CommandLine(command)
               .setCaseInsensitiveEnumValuesAllowed(true)
               .execute(args);
+
+        System.exit(exitCode);
     }
 }
diff --git a/src/main/java/io/github/s7i/doer/Globals.java b/src/main/java/io/github/s7i/doer/Globals.java
index e4017d1..d98489e 100644
--- a/src/main/java/io/github/s7i/doer/Globals.java
+++ b/src/main/java/io/github/s7i/doer/Globals.java
@@ -3,20 +3,23 @@
 import io.github.s7i.doer.domain.ConfigProcessor;
 import io.github.s7i.doer.domain.kafka.KafkaFactory;
 import io.github.s7i.doer.domain.output.OutputFactory;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Supplier;
+import io.github.s7i.doer.pipeline.Pipeline;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+/**
+ * <pre>
+ *     "Dependency injection is a kind of globals with configuration."
+ * </pre>
+ */
 @Slf4j
 public enum Globals implements Context {
     INSTANCE;
@@ -26,6 +29,8 @@ public enum Globals implements Context {
     List<Runnable> stopHooks = new ArrayList<>();
     @Getter
     KafkaFactory kafka = new KafkaFactory();
+    @Getter
+    Pipeline pipeline = new Pipeline();
 
     public Scope getScope() {
 
diff --git a/src/main/java/io/github/s7i/doer/command/CommandWithContext.java b/src/main/java/io/github/s7i/doer/command/CommandWithContext.java
new file mode 100644
index 0000000..d8c61ff
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/command/CommandWithContext.java
@@ -0,0 +1,25 @@
+package io.github.s7i.doer.command;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.config.Base;
+
+import static java.util.Objects.requireNonNull;
+
+public abstract class CommandWithContext<M extends Base> extends ManifestFileCommand {
+
+    protected abstract Class<M> manifestClass();
+
+    @Override
+    public void onExecuteCommand() {
+        var manifest = requireNonNull(parseYaml(manifestClass()), "manifest definition");
+
+        var ctx = new Context.Initializer(Context.InitialParameters.builder()
+                .workDir(yaml.toPath().toAbsolutePath().getParent())
+                .params(manifest.getParams())
+                .build())
+                .context();
+        onExecuteCommand(ctx, manifest);
+    }
+
+    public abstract void onExecuteCommand(Context context, M manifest);
+}
diff --git a/src/main/java/io/github/s7i/doer/command/Ingest.java b/src/main/java/io/github/s7i/doer/command/Ingest.java
new file mode 100644
index 0000000..6276a2b
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/command/Ingest.java
@@ -0,0 +1,27 @@
+package io.github.s7i.doer.command;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.domain.ingest.IngestProcessor;
+import io.github.s7i.doer.manifest.ingest.IngestRecordManifest;
+import picocli.CommandLine.Command;
+
+import java.io.File;
+
+@Command(name = "ingest")
+public class Ingest extends CommandWithContext<IngestRecordManifest> {
+
+    @Override
+    protected File getDefaultManifestFile() {
+        return new File("ingest.yml");
+    }
+
+    @Override
+    protected Class<IngestRecordManifest> manifestClass() {
+        return IngestRecordManifest.class;
+    }
+
+    @Override
+    public void onExecuteCommand(Context ctx, IngestRecordManifest manifest) {
+        new IngestProcessor(ctx).process(manifest);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/command/SinkCommand.java b/src/main/java/io/github/s7i/doer/command/SinkCommand.java
new file mode 100644
index 0000000..1cf958a
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/command/SinkCommand.java
@@ -0,0 +1,31 @@
+package io.github.s7i.doer.command;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.domain.SinkProcessor;
+import io.github.s7i.doer.manifest.SinkManifest;
+import picocli.CommandLine;
+
+import java.io.File;
+import java.util.Collections;
+
+import static java.util.Objects.isNull;
+
+@CommandLine.Command(name = "sink")
+public class SinkCommand extends CommandWithContext<SinkManifest> {
+
+    @Override
+    protected File getDefaultManifestFile() {
+        return new File("sink.yml");
+    }
+
+    @Override
+    protected Class<SinkManifest> manifestClass() {
+        return SinkManifest.class;
+    }
+
+    @Override
+    public void onExecuteCommand(Context context, SinkManifest manifest) {
+        var list = manifest.getSpec();
+        new SinkProcessor(context).execute(isNull(list) ? Collections.emptyList() : list);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/command/util/CommandManifest.java b/src/main/java/io/github/s7i/doer/command/util/CommandManifest.java
index 82b61d0..b2cd808 100644
--- a/src/main/java/io/github/s7i/doer/command/util/CommandManifest.java
+++ b/src/main/java/io/github/s7i/doer/command/util/CommandManifest.java
@@ -1,24 +1,35 @@
 package io.github.s7i.doer.command.util;
 
+import io.github.s7i.doer.Doer;
+import io.github.s7i.doer.Globals;
 import io.github.s7i.doer.command.Command;
+import io.github.s7i.doer.command.Ingest;
 import io.github.s7i.doer.command.KafkaFeeder;
+import io.github.s7i.doer.command.SinkCommand;
 import io.github.s7i.doer.command.dump.KafkaDump;
+import io.github.s7i.doer.domain.ConfigProcessor;
+import io.github.s7i.doer.util.Banner;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import picocli.CommandLine;
+import picocli.CommandLine.Option;
+import picocli.CommandLine.Parameters;
+
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
-import io.github.s7i.doer.util.Banner;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import picocli.CommandLine;
-import picocli.CommandLine.Option;
-import picocli.CommandLine.Parameters;
-
+import static io.github.s7i.doer.Doer.DOER_CONSOLE;
 import static io.github.s7i.doer.command.ManifestFileCommand.Builder.fromManifestFile;
 import static java.util.Objects.requireNonNull;
 
@@ -27,8 +38,8 @@
         aliases = "cm",
         description = "Parsing command manifest yaml file."
 )
-@Slf4j
-public class CommandManifest implements Runnable, Banner {
+@Slf4j(topic = DOER_CONSOLE)
+public class CommandManifest implements Callable<Integer>, Banner {
 
     public static final int MAX_THR_COUNT = 20;
     @Parameters(arity = "1..*")
@@ -48,6 +59,9 @@ static class TaskWrapper implements Runnable {
 
         final String name;
         final Command coreTask;
+        Duration duration;
+        @Getter
+        boolean failed;
 
         @Override
         public void run() {
@@ -55,53 +69,101 @@ public void run() {
             var begin = Instant.now();
             try {
                 try {
-                    coreTask.call();
+                    failed = 0 != coreTask.call();
                 } catch (Exception e) {
-                    log.error("fatal error", e);
+                    log.error("[TASK FAILURE]", e);
+                    failed = true;
                 }
             } finally {
-                var duration = Duration.between(begin, Instant.now());
-                log.info("Task {} ends in {}", name, duration);
+                duration = Duration.between(begin, Instant.now());
             }
         }
+
+        public String getSummary() {
+            return String.format("%sTask %s ends in %s", failed ? "[FAILED] " : "", name, duration);
+        }
+    }
+
+    private Optional<Runnable> mapToTask(File yaml) {
+        try (var br = Files.newBufferedReader(yaml.toPath())) {
+            String version = br.readLine();
+            String kind = br.readLine();
+            kind = kind.substring(kind.lastIndexOf(':') + 1).trim();
+
+            switch (kind) {
+                case "kafka-ingest":
+                    return Optional.of(new TaskWrapper(kind, fromManifestFile(KafkaFeeder.class, yaml)));
+                case "kafka-dump":
+                    return Optional.of(new TaskWrapper(kind, fromManifestFile(KafkaDump.class, yaml)));
+                case "ingest":
+                    return Optional.of(new TaskWrapper(kind, fromManifestFile(Ingest.class, yaml)));
+                case "sink":
+                    return Optional.of(new TaskWrapper(kind, fromManifestFile(SinkCommand.class, yaml)));
+                case "config":
+                    return Optional.of(() -> new ConfigProcessor(yaml).processConfig());
+                default:
+                    log.warn("unsupported kind/type of file: {}", kind);
+                    break;
+            }
+
+        } catch (IOException e) {
+            log.error("reading yaml file", e);
+        }
+        return Optional.empty();
     }
 
     @Override
-    public void run() {
+    public Integer call() {
         printBanner();
         requireNonNull(yamls, "manifest file set...");
 
-        var pool = Executors.newFixedThreadPool(Math.min(yamls.length, MAX_THR_COUNT), this::spawnNewThread);
-        for (var yaml : yamls) {
-            try (var br = Files.newBufferedReader(yaml.toPath())) {
-                String version = br.readLine();
-                String kind = br.readLine();
-                kind = kind.substring(kind.lastIndexOf(':') + 1).trim();
-
-                switch (kind) {
-                    case "kafka-ingest":
-                        pool.execute(new TaskWrapper(kind, fromManifestFile(KafkaFeeder.class, yaml)));
-                        break;
-                    case "kafka-dump":
-                        pool.execute(new TaskWrapper(kind, fromManifestFile(KafkaDump.class, yaml)));
-                        break;
-                    default:
-                        log.warn("unsupported kind/type of file: {}", kind);
-                        break;
-
-                }
+        var tasks = Stream.of(yamls)
+                .map(this::mapToTask)
+                .flatMap(Optional::stream)
+                .collect(Collectors.toList());
 
-            } catch (IOException e) {
-                log.error("reading yaml file", e);
-            }
+        if (tasks.isEmpty()) {
+            log.info("Nothing to run...");
+            return Doer.EC_INVALID_USAGE;
         }
+
+        var pool = Executors.newFixedThreadPool(Math.min(tasks.size(), MAX_THR_COUNT), this::spawnNewThread);
+        tasks.forEach(pool::execute);
+
         try {
             pool.shutdown();
-            pool.awaitTermination(24, TimeUnit.HOURS);
-            log.info("All task ended");
+            if(!pool.awaitTermination(24, TimeUnit.HOURS)) {
+                log.warn("Timeout...");
+            }
+
+            log.info("All task finished.");
+            log.info("Summary:");
+
+            var sum = tasks.stream()
+                    .filter(TaskWrapper.class::isInstance)
+                    .mapToLong(t -> {
+                        var task = (TaskWrapper) t;
+                        log.info(task.getSummary());
+                        return task.duration.toNanos();
+                    })
+                    .sum();
+
+            log.info("Total time: {}.", Duration.ofNanos(sum));
+
+            Globals.INSTANCE.stopAllSilent();
+
+            var hasFailed = tasks.stream()
+                    .filter(TaskWrapper.class::isInstance)
+                    .anyMatch(t -> ((TaskWrapper) t).isFailed());
+
+            if (hasFailed) {
+                return Doer.EC_ERROR;
+            }
+
         } catch (InterruptedException e) {
-            log.error("", e);
+            log.error("oops", e);
             Thread.currentThread().interrupt();
         }
+        return 0;
     }
 }
diff --git a/src/main/java/io/github/s7i/doer/command/util/Misc.java b/src/main/java/io/github/s7i/doer/command/util/Misc.java
index 69c0edf..3bad984 100644
--- a/src/main/java/io/github/s7i/doer/command/util/Misc.java
+++ b/src/main/java/io/github/s7i/doer/command/util/Misc.java
@@ -6,7 +6,6 @@
 import io.github.s7i.doer.HandledRuntimeException;
 import io.github.s7i.doer.command.file.ReplaceInFile;
 import io.github.s7i.doer.domain.output.HttpOutput;
-import io.github.s7i.doer.pipeline.PipelineService;
 import io.github.s7i.doer.util.Clipboard;
 import io.github.s7i.doer.util.GitProps;
 import io.github.s7i.doer.util.PropertyResolver;
@@ -23,8 +22,6 @@
 import org.jeasy.rules.support.reader.YamlRuleDefinitionReader;
 import org.mvel2.MVEL;
 import org.mvel2.compiler.CompiledExpression;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
@@ -52,14 +49,12 @@
         description = "Miscellaneous command set.",
         subcommands = {
                 ReplaceInFile.class,
-                CommandManifest.class,
-                PipelineService.class
+                CommandManifest.class
         }
 )
 @Slf4j(topic = "doer.console")
 public class Misc {
 
-    public static final Logger LOG = LoggerFactory.getLogger(Misc.class);
 
     @Command
     public void time(@Parameters(paramLabel = "time-value") long time,
@@ -240,7 +235,7 @@ public void emitLogToElastic(
                                 try {
                                     return nonNull(logFile) ? Files.newBufferedReader(logFile) : null;
                                 } catch (IOException e) {
-                                    LOG.error("log file {}", logFile, e);
+                                    log.error("log file {}", logFile, e);
                                     return null;
                                 }
                             },
@@ -253,7 +248,7 @@ public void emitLogToElastic(
                                         return new BufferedReader(new InputStreamReader(is));
                                     }
                                 } catch (Exception e) {
-                                    LOG.error("exec {}", exec, e);
+                                    log.error("exec {}", exec, e);
 
                                 }
                                 return null;
@@ -261,10 +256,10 @@ public void emitLogToElastic(
                             () -> {
                                 try {
                                     var clipboard = Clipboard.getString();
-                                    LOG.debug("clipboard data: {}", clipboard);
+                                    log.debug("clipboard data: {}", clipboard);
                                     return new BufferedReader(new StringReader(clipboard));
                                 } catch (Exception e) {
-                                    LOG.error("using clipboard", e);
+                                    log.error("using clipboard", e);
                                     return null;
                                 }
                             })
diff --git a/src/main/java/io/github/s7i/doer/domain/Mappers.java b/src/main/java/io/github/s7i/doer/domain/Mappers.java
new file mode 100644
index 0000000..8027fb4
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/Mappers.java
@@ -0,0 +1,73 @@
+package io.github.s7i.doer.domain;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+import io.github.s7i.doer.DoerException;
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.pipeline.proto.PipelineLoad;
+import io.github.s7i.doer.proto.Record;
+import lombok.experimental.UtilityClass;
+
+import java.util.HashMap;
+
+import static java.util.Objects.nonNull;
+
+@UtilityClass
+public class Mappers {
+
+    public static Output.Load mapFrom(PipelineLoad value) throws InvalidProtocolBufferException {
+        if (!value.getLoad().is(Record.class)) {
+            throw new DoerException("invalid load type: " + value.getLoad().getTypeUrl());
+        }
+        var rec = value.getLoad().unpack(Record.class);
+
+        var bld = Output.Load.builder();
+
+        if (rec.hasKey()) {
+            bld.key(rec.getKey().getValue());
+        }
+
+        if (rec.hasResource()) {
+            bld.resource(rec.getResource().getValue());
+        }
+
+        if (rec.hasData()) {
+            var recordData = rec.getData().getValue();
+            bld.data(recordData.toByteArray());
+        }
+
+        if (!rec.getMetaMap().isEmpty()) {
+            bld.meta(new HashMap<>(rec.getMetaMap()));
+        }
+
+        return bld.build();
+
+    }
+
+    public static Record mapFrom(Output.Load load) {
+        var b = Record.newBuilder();
+        if (nonNull(load.getResource())) {
+            b.setResource(StringValue.of(load.getResource()));
+        }
+        if (nonNull(load.getKey())) {
+            b.setKey(StringValue.of(load.getKey()));
+        }
+        if (nonNull(load.getMeta())) {
+            b.putAllMeta(load.getMeta());
+        }
+        var data = load.getData();
+        if (nonNull(data) && data.length > 0) {
+            b.setData(BytesValue.of(ByteString.copyFrom(data)));
+        }
+        return b.build();
+    }
+
+    public static Output.Load mapFrom(String resource, byte[] data) {
+        return Output.Load.builder()
+                .resource(resource)
+                .data(data)
+                .build();
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/ServiceEntrypoint.java b/src/main/java/io/github/s7i/doer/domain/ServiceEntrypoint.java
new file mode 100644
index 0000000..0b24f4a
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/ServiceEntrypoint.java
@@ -0,0 +1,73 @@
+package io.github.s7i.doer.domain;
+
+import com.google.protobuf.Empty;
+import io.github.s7i.doer.proto.DoerServiceGrpc;
+import io.github.s7i.doer.proto.Sone;
+import io.github.s7i.doer.util.GitProps;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+@Slf4j
+public class ServiceEntrypoint extends DoerServiceGrpc.DoerServiceImplBase {
+
+    static class Pulsar {
+        ExecutorService es = Executors.newFixedThreadPool(1);
+        List<Consumer<String>> receivers = new CopyOnWriteArrayList<>();
+
+        void begin() {
+            es.submit(this::pulse);
+
+        }
+
+        void end() {
+            es.shutdown();
+            receivers.clear();
+        }
+
+        void makePulse(Consumer<String> p) {
+            final var now = Instant.now();
+
+            p.accept(String.format("%d|%s", now.getEpochSecond() ,now));
+        }
+
+        void pulse() {
+            while (!es.isShutdown()) {
+                try {
+                    receivers.forEach(this::makePulse);
+                    TimeUnit.SECONDS.sleep(1);
+                } catch (InterruptedException e) {
+                    log.error("oops", e);
+                    Thread.currentThread().interrupt();
+                }
+            }
+        }
+
+        void add(Consumer<String> receiver) {
+            receivers.add(receiver);
+        }
+    }
+
+    @Override
+    public void version(Empty request, StreamObserver<Sone> responseObserver) {
+        responseObserver.onNext(sone(new GitProps().toString()));
+        responseObserver.onCompleted();
+    }
+
+
+    @Override
+    public void pulse(Empty request, StreamObserver<Sone> responseObserver) {
+        super.pulse(request, responseObserver);
+    }
+
+    public Sone sone(String s) {
+        return Sone.newBuilder().setS(s).build();
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/SinkProcessor.java b/src/main/java/io/github/s7i/doer/domain/SinkProcessor.java
new file mode 100644
index 0000000..8bee1d6
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/SinkProcessor.java
@@ -0,0 +1,125 @@
+package io.github.s7i.doer.domain;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.DoerException;
+import io.github.s7i.doer.Globals;
+import io.github.s7i.doer.domain.output.DefaultOutputProvider;
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.manifest.SinkManifest;
+import io.github.s7i.doer.pipeline.PipePuller;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static java.util.Objects.nonNull;
+
+@RequiredArgsConstructor
+@Slf4j
+public class SinkProcessor implements DefaultOutputProvider {
+
+
+    final Context context;
+    ExecutorService executorService;
+    CountDownLatch taskToComplete;
+
+    transient int failureCount;
+
+    public void execute(List<SinkManifest.SinkSpec> specList) {
+
+        context.lookupPipeline().ifPresent(pipeline -> {
+            var puller = pipeline.connect().lookupPuller();
+
+            var enabled = specList.stream()
+                    .filter(SinkManifest.SinkSpec::isEnabled)
+                    .collect(Collectors.toList());
+
+            if (!enabled.isEmpty()) {
+                enableOutputs(enabled, puller);
+            } else {
+                drainPipeline(puller);
+            }
+        });
+        if (nonNull(taskToComplete)) {
+            log.debug("Awaiting to complete all tasks...");
+
+            try {
+                taskToComplete.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new DoerException(e);
+            }
+            try {
+                executorService.shutdown();
+                if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
+                    log.warn("Some timeouts...");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new DoerException(e);
+            }
+            log.debug("All tasks done.");
+        }
+        if (failureCount > 0) {
+            throw new DoerException("Some failures during executions.");
+        }
+    }
+
+    private void enableOutputs(List<SinkManifest.SinkSpec> specList, PipePuller puller) {
+        var initial = context.getParams();
+        var size = specList.size();
+
+        log.debug("Number of output tasks: {}", size);
+
+        taskToComplete = new CountDownLatch(size);
+
+        var tg = new ThreadGroup("doer-output-workers");
+        executorService = Executors.newFixedThreadPool(size, runnable ->
+                new Thread(tg, runnable, "output-worker"));
+
+        specList.stream()
+                .map(s -> toRunnable(s, puller, initial))
+                .forEach(executorService::submit);
+    }
+
+    private Runnable toRunnable(SinkManifest.SinkSpec spec, PipePuller puller, Map<String, String> param) {
+        return () -> {
+            try {
+                Globals.INSTANCE.shareParams(param);
+
+                loopOutput(context.buildOutput(spec::getOutput), puller);
+            } catch (RuntimeException r) {
+                log.error("worker task issue", r);
+                failureCount++;
+            } finally {
+                log.debug("task completed {}", Thread.currentThread().getName());
+                taskToComplete.countDown();
+            }
+        };
+    }
+
+    private void drainPipeline(PipePuller puller) {
+        log.debug("Using default output.");
+
+        var out = getDefaultOutput(context);
+        loopOutput(out, puller);
+    }
+
+    public static void loopOutput(Output out, PipePuller puller) {
+        Output.Load load;
+        while (null != (load = puller.onNextLoad())) {
+            out.emit(load);
+
+            puller.onAccept();
+        }
+        LoggerFactory.getLogger(SinkManifest.class).debug("output loop ends.");
+    }
+
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/grpc/GrpcServer.java b/src/main/java/io/github/s7i/doer/domain/grpc/GrpcServer.java
new file mode 100644
index 0000000..74f983d
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/grpc/GrpcServer.java
@@ -0,0 +1,54 @@
+package io.github.s7i.doer.domain.grpc;
+
+import io.grpc.BindableService;
+import io.grpc.Grpc;
+import io.grpc.InsecureServerCredentials;
+import io.grpc.Server;
+import io.grpc.protobuf.services.HealthStatusManager;
+import io.grpc.protobuf.services.ProtoReflectionService;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j(topic = "doer.console")
+@RequiredArgsConstructor
+public class GrpcServer {
+
+    private final HealthStatusManager hsm = new HealthStatusManager();
+    private Server server;
+
+    private final int port;
+    private final BindableService service;
+
+    public GrpcServer startServer() throws IOException {
+
+        server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
+                .addService(service)
+                .addService(ProtoReflectionService.newInstance())
+                .addService(hsm.getHealthService())
+                .build()
+                .start();
+
+        log.info("Server Started: {}", server);
+
+        Runtime.getRuntime().addShutdownHook(new Thread(this::stopServer, "grpc-server-shutdown-hook"));
+
+        try {
+            server.awaitTermination();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("oops", e);
+        }
+        log.info("bye bye");
+
+        return this;
+    }
+
+    @SneakyThrows
+    private void stopServer() {
+        server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/ingest/IngestProcessor.java b/src/main/java/io/github/s7i/doer/domain/ingest/IngestProcessor.java
new file mode 100644
index 0000000..004af80
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/ingest/IngestProcessor.java
@@ -0,0 +1,50 @@
+package io.github.s7i.doer.domain.ingest;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.DoerException;
+import io.github.s7i.doer.domain.output.DefaultOutputProvider;
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.manifest.ingest.IngestRecordManifest;
+import io.github.s7i.doer.util.PropertyResolver;
+import lombok.RequiredArgsConstructor;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Optional;
+
+@RequiredArgsConstructor
+public class IngestProcessor implements DefaultOutputProvider {
+
+    private final Context context;
+    private PropertyResolver propertyResolver;
+
+    public void process(IngestRecordManifest manifest) {
+        propertyResolver = context.getPropertyResolver();
+
+        try (var output = getDefaultOutput(context)) {
+
+            manifest.getRecords().stream()
+                    .map(this::resolveRecord)
+                    .forEach(output::emit);
+        } catch (Exception e) {
+            throw new DoerException(e);
+        }
+    }
+
+    private Output.Load resolveRecord(IngestRecordManifest.Record record) {
+        final var bld = Output.Load.builder();
+
+        Optional.ofNullable(record.getRecord()).ifPresent(data ->
+                bld.data(propertyResolver
+                        .resolve(record.getRecord())
+                        .getBytes(StandardCharsets.UTF_8))
+        );
+
+        Optional.ofNullable(record.getKey()).ifPresent(key ->
+                bld.key(propertyResolver.resolve(record.getKey()))
+        );
+        bld.meta(Map.of("doer.record.type", "ingest-record"));
+
+        return bld.build();
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/kafka/output/KafkaOutput.java b/src/main/java/io/github/s7i/doer/domain/kafka/output/KafkaOutput.java
index 9382700..7d820d5 100644
--- a/src/main/java/io/github/s7i/doer/domain/kafka/output/KafkaOutput.java
+++ b/src/main/java/io/github/s7i/doer/domain/kafka/output/KafkaOutput.java
@@ -1,38 +1,65 @@
 package io.github.s7i.doer.domain.kafka.output;
 
-import static java.util.Objects.nonNull;
-
+import io.github.s7i.doer.DoerException;
 import io.github.s7i.doer.domain.kafka.KafkaConfig;
 import io.github.s7i.doer.domain.kafka.KafkaFactory;
 import io.github.s7i.doer.domain.output.Output;
-import java.nio.charset.StandardCharsets;
 import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.internals.RecordHeader;
 
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.util.Objects.nonNull;
+import static java.util.Objects.requireNonNull;
+
 @Builder
+@Slf4j
 public class KafkaOutput implements Output {
 
+    public static final int RECORD_TIMEOUT_SEC = 30;
     final KafkaFactory kafkaFactory;
     final KafkaConfig config;
     final boolean userTracing;
     final TopicPartition topic;
     Producer<String, byte[]> producer;
 
+    boolean closed;
+
     @Override
     public void open() {
+        if (nonNull(producer)) {
+            return;
+        }
         producer = kafkaFactory.getProducerFactory().createProducer(config, userTracing);
     }
 
     @Override
     public void emit(Load load) {
+        if (closed) {
+            throw new DoerException("closed");
+        }
+        requireNonNull(producer, "producer");
         var record = new ProducerRecord<>(topic.topic(), load.getKey(), load.getData());
         if (nonNull(load.getMeta())) {
             load.getMeta().forEach((k, v) -> record.headers().add(new RecordHeader(k, v.getBytes(StandardCharsets.UTF_8))));
         }
-        producer.send(record);
+        log.debug("sending record...");
+        try {
+            producer.send(record, (m, t) -> log.debug("rec-callback: {}", m, t))
+                    .get(RECORD_TIMEOUT_SEC, TimeUnit.SECONDS);
+        } catch (InterruptedException itr) {
+            Thread.currentThread().interrupt();
+            throw new DoerException("interrupted thread");
+        } catch (ExecutionException | TimeoutException e) {
+            throw new DoerException(e);
+        }
     }
 
     @Override
@@ -40,5 +67,7 @@ public void close() throws Exception {
         if (nonNull(producer)) {
             producer.close();
         }
+        closed = true;
+        log.debug("closed");
     }
 }
diff --git a/src/main/java/io/github/s7i/doer/domain/output/DefaultOutputProvider.java b/src/main/java/io/github/s7i/doer/domain/output/DefaultOutputProvider.java
new file mode 100644
index 0000000..2b44829
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/output/DefaultOutputProvider.java
@@ -0,0 +1,17 @@
+package io.github.s7i.doer.domain.output;
+
+import io.github.s7i.doer.Context;
+import io.github.s7i.doer.util.Mark;
+
+public interface DefaultOutputProvider {
+    @Mark.Param
+    String DOER_OUTPUT = "doer.output";
+
+    default Output getDefaultOutput(Context context) {
+        return context.getParams().entrySet().stream()
+                .filter(e -> e.getKey().equals(DOER_OUTPUT))
+                .map(def -> context.buildOutput(def::getValue))
+                .findFirst()
+                .orElseGet(ConsoleOutput::new);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/output/HttpOutput.java b/src/main/java/io/github/s7i/doer/domain/output/HttpOutput.java
index c0d84ea..d6ef071 100644
--- a/src/main/java/io/github/s7i/doer/domain/output/HttpOutput.java
+++ b/src/main/java/io/github/s7i/doer/domain/output/HttpOutput.java
@@ -1,5 +1,8 @@
 package io.github.s7i.doer.domain.output;
 
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.IOException;
 import java.net.URI;
 import java.net.http.HttpClient;
@@ -7,8 +10,8 @@
 import java.net.http.HttpRequest.BodyPublishers;
 import java.net.http.HttpResponse.BodyHandlers;
 import java.time.Duration;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
+
+import static java.util.Objects.nonNull;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -19,6 +22,9 @@ public class HttpOutput implements Output {
 
     @Override
     public void open() {
+        if (nonNull(httpClient)) {
+            return;
+        }
         httpClient = HttpClient.newBuilder()
               .connectTimeout(Duration.ofSeconds(15))
               .build();
diff --git a/src/main/java/io/github/s7i/doer/domain/output/Output.java b/src/main/java/io/github/s7i/doer/domain/output/Output.java
index 6a75eaa..bd10193 100644
--- a/src/main/java/io/github/s7i/doer/domain/output/Output.java
+++ b/src/main/java/io/github/s7i/doer/domain/output/Output.java
@@ -1,16 +1,13 @@
 package io.github.s7i.doer.domain.output;
 
-import static java.util.Objects.nonNull;
-
-import com.google.protobuf.ByteString;
-import com.google.protobuf.BytesValue;
-import com.google.protobuf.StringValue;
+import io.github.s7i.doer.domain.Mappers;
 import io.github.s7i.doer.proto.Record;
-import java.util.Map;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 
+import java.util.Map;
+
 public interface Output extends AutoCloseable {
 
     @Builder
@@ -28,30 +25,14 @@ public String dataAsString() {
         }
 
         public Record toRecord() {
-            var b = Record.newBuilder();
-            if (nonNull(resource)) {
-                b.setResource(StringValue.of(resource));
-            }
-            if (nonNull(key)) {
-                b.setKey(StringValue.of(key));
-            }
-            if (nonNull(meta)) {
-                b.putAllMeta(meta);
-            }
-            if (nonNull(data) && data.length > 0) {
-                b.setData(BytesValue.of(ByteString.copyFrom(data)));
-            }
-            return b.build();
+            return Mappers.mapFrom(this);
         }
     }
 
     void open();
 
     default void emit(String resource, byte[] data) {
-        emit(Load.builder()
-              .resource(resource)
-              .data(data)
-              .build());
+        emit(Mappers.mapFrom(resource, data));
     }
 
     void emit(Load load);
diff --git a/src/main/java/io/github/s7i/doer/domain/output/OutputBuilder.java b/src/main/java/io/github/s7i/doer/domain/output/OutputBuilder.java
index aa8629c..fba3d01 100644
--- a/src/main/java/io/github/s7i/doer/domain/output/OutputBuilder.java
+++ b/src/main/java/io/github/s7i/doer/domain/output/OutputBuilder.java
@@ -1,29 +1,43 @@
 package io.github.s7i.doer.domain.output;
 
 import io.github.s7i.doer.Context;
+import io.github.s7i.doer.Globals;
 import io.github.s7i.doer.domain.kafka.output.KafkaOutputCreator;
 import io.github.s7i.doer.domain.kafka.output.KafkaUri;
 import io.github.s7i.doer.domain.output.creator.FileOutputCreator;
 import io.github.s7i.doer.domain.output.creator.HttpOutputCreator;
+import io.github.s7i.doer.domain.output.creator.PipelineOutputCreator;
 import lombok.Setter;
 import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
 
 @Accessors(fluent = true)
 @Setter
+@Slf4j
 public class OutputBuilder {
     Context context;
 
     public Output build(OutputProvider outputProvider) {
-        FileOutputCreator foc = () -> context.getBaseDir().resolve(outputProvider.getOutput());
+        var def = outputProvider.getOutput();
+
+        FileOutputCreator foc = () -> context.getBaseDir().resolve(def);
         HttpOutputCreator http = outputProvider::getOutput;
         KafkaOutputCreator kafka = new KafkaUri(outputProvider, context);
+        PipelineOutputCreator pipeline = () -> Globals.INSTANCE.getPipeline().connect();
+
 
         final var factory = context.getOutputFactory();
         factory.register(OutputKind.FILE, foc);
         factory.register(OutputKind.HTTP, http);
         factory.register(OutputKind.KAFKA, kafka);
+        factory.register(OutputKind.PIPELINE, pipeline);
 
-        return factory.resolve(new UriResolver(outputProvider.getOutput()))
-                .orElseThrow();
+        var output = factory.resolve(new UriResolver(def)).orElseThrow();
+        log.debug("resolved output {}", output);
+        //TODO: auto open make configurable
+        output.open();
+
+        return output;
     }
+
 }
diff --git a/src/main/java/io/github/s7i/doer/domain/output/OutputKind.java b/src/main/java/io/github/s7i/doer/domain/output/OutputKind.java
index fc354af..1c1c735 100644
--- a/src/main/java/io/github/s7i/doer/domain/output/OutputKind.java
+++ b/src/main/java/io/github/s7i/doer/domain/output/OutputKind.java
@@ -2,5 +2,5 @@
 
 public enum OutputKind {
 
-    CONSOLE, FILE, KAFKA, HTTP
+    CONSOLE, FILE, KAFKA, HTTP, PIPELINE
 }
diff --git a/src/main/java/io/github/s7i/doer/domain/output/PipelineOutput.java b/src/main/java/io/github/s7i/doer/domain/output/PipelineOutput.java
new file mode 100644
index 0000000..f0ad9fd
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/output/PipelineOutput.java
@@ -0,0 +1,36 @@
+package io.github.s7i.doer.domain.output;
+
+import io.github.s7i.doer.pipeline.BlockingPipe;
+import io.github.s7i.doer.pipeline.PipeConnection;
+import lombok.RequiredArgsConstructor;
+
+import static java.util.Objects.nonNull;
+import static java.util.Objects.requireNonNull;
+
+@RequiredArgsConstructor
+public class PipelineOutput implements Output {
+
+
+    private final PipeConnection pipeConnection;
+    private BlockingPipe pusher;
+
+    @Override
+    public void open() {
+        if (nonNull(pusher)) {
+            return;
+        }
+        pusher = new BlockingPipe();
+        pipeConnection.registerPuller(pusher);
+    }
+
+    @Override
+    public void emit(Load load) {
+        requireNonNull(pusher);
+        pusher.offer(load);
+    }
+
+    @Override
+    public void close() throws Exception {
+
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/domain/output/UriResolver.java b/src/main/java/io/github/s7i/doer/domain/output/UriResolver.java
index 06e77fa..4a32847 100644
--- a/src/main/java/io/github/s7i/doer/domain/output/UriResolver.java
+++ b/src/main/java/io/github/s7i/doer/domain/output/UriResolver.java
@@ -1,6 +1,7 @@
 package io.github.s7i.doer.domain.output;
 
 import io.github.s7i.doer.DoerException;
+
 import java.net.URI;
 import java.net.URISyntaxException;
 
@@ -28,6 +29,8 @@ public OutputKind resolveOutputKind() {
                 return OutputKind.valueOf(authority.toUpperCase());
             case "kafka":
                 return OutputKind.KAFKA;
+            case "pipeline":
+                return OutputKind.PIPELINE;
             default:
                 return OutputKind.FILE;
         }
diff --git a/src/main/java/io/github/s7i/doer/domain/output/creator/PipelineOutputCreator.java b/src/main/java/io/github/s7i/doer/domain/output/creator/PipelineOutputCreator.java
new file mode 100644
index 0000000..2ce7ce4
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/domain/output/creator/PipelineOutputCreator.java
@@ -0,0 +1,18 @@
+package io.github.s7i.doer.domain.output.creator;
+
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.domain.output.PipelineOutput;
+import io.github.s7i.doer.pipeline.PipeConnection;
+
+public interface PipelineOutputCreator extends OutputCreator {
+
+
+    PipeConnection getLoadPipe();
+
+    @Override
+    default Output create() {
+        var pipelineOutput = new PipelineOutput(getLoadPipe());
+        pipelineOutput.open();
+        return pipelineOutput;
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/manifest/SinkManifest.java b/src/main/java/io/github/s7i/doer/manifest/SinkManifest.java
new file mode 100644
index 0000000..e00483e
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/manifest/SinkManifest.java
@@ -0,0 +1,20 @@
+package io.github.s7i.doer.manifest;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.github.s7i.doer.config.Base;
+import lombok.Data;
+import lombok.Getter;
+
+import java.util.List;
+
+@Getter
+public class SinkManifest extends Base {
+    @Data
+    public static class SinkSpec {
+        String output = "";
+        boolean enabled = true;
+    }
+
+    @JsonProperty("spec")
+    List<SinkSpec> spec;
+}
diff --git a/src/main/java/io/github/s7i/doer/manifest/ingest/IngestRecordManifest.java b/src/main/java/io/github/s7i/doer/manifest/ingest/IngestRecordManifest.java
new file mode 100644
index 0000000..b8d80e9
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/manifest/ingest/IngestRecordManifest.java
@@ -0,0 +1,30 @@
+package io.github.s7i.doer.manifest.ingest;
+
+import com.fasterxml.jackson.annotation.JsonAlias;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.github.s7i.doer.config.Base;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.util.Collections;
+import java.util.List;
+
+@Getter
+public class IngestRecordManifest extends Base {
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class Record {
+        String record;
+        String key;
+    }
+
+    @JsonProperty("ingest")
+    @JsonAlias("spec")
+    List<Record> records = Collections.emptyList();
+
+
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/BackendFactory.java b/src/main/java/io/github/s7i/doer/pipeline/BackendFactory.java
new file mode 100644
index 0000000..bc898b5
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/BackendFactory.java
@@ -0,0 +1,7 @@
+package io.github.s7i.doer.pipeline;
+
+public interface BackendFactory {
+
+    PipeConnection create();
+
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/BlockingPipe.java b/src/main/java/io/github/s7i/doer/pipeline/BlockingPipe.java
new file mode 100644
index 0000000..2c72334
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/BlockingPipe.java
@@ -0,0 +1,63 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.domain.output.Output;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class BlockingPipe implements PipePuller, PipePusher {
+
+    public static final int SLEEP_FOR_CHANGE = 10;
+    private final ArrayBlockingQueue<Output.Load> queue = new ArrayBlockingQueue<>(1);
+
+    /**
+     * Blocking call.
+     *
+     * @return
+     */
+    @Override
+    public Output.Load onNextLoad() {
+
+        Output.Load load;
+        while ((load = queue.peek()) == null && !Thread.currentThread().isInterrupted()) {
+            try {
+                TimeUnit.MILLISECONDS.sleep(SLEEP_FOR_CHANGE);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+        }
+
+        return load;
+    }
+
+    @Override
+    public void onAccept() {
+        queue.poll();
+    }
+
+
+    /**
+     * Blocking call.
+     *
+     * @param load
+     */
+    @Override
+    public void offer(Output.Load load) {
+        boolean accepted;
+        do {
+            accepted = queue.offer(load);
+            if (!accepted) {
+                try {
+                    TimeUnit.MILLISECONDS.sleep(SLEEP_FOR_CHANGE);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
+        while (!accepted);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/PipeConnection.java b/src/main/java/io/github/s7i/doer/pipeline/PipeConnection.java
new file mode 100644
index 0000000..254fb7d
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/PipeConnection.java
@@ -0,0 +1,14 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.DoerException;
+
+public interface PipeConnection {
+
+    default void registerPuller(PipePuller puller) {
+        throw new DoerException("not implemented");
+    }
+
+    default PipePuller lookupPuller() {
+        throw new DoerException("not implemented");
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/PipePuller.java b/src/main/java/io/github/s7i/doer/pipeline/PipePuller.java
new file mode 100644
index 0000000..57dca56
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/PipePuller.java
@@ -0,0 +1,10 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.domain.output.Output;
+
+public interface PipePuller {
+
+    Output.Load onNextLoad();
+
+    void onAccept();
+}
\ No newline at end of file
diff --git a/src/main/java/io/github/s7i/doer/pipeline/PipePusher.java b/src/main/java/io/github/s7i/doer/pipeline/PipePusher.java
new file mode 100644
index 0000000..a6a238e
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/PipePusher.java
@@ -0,0 +1,8 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.domain.output.Output;
+
+public interface PipePusher {
+
+    void offer(Output.Load load);
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/Pipeline.java b/src/main/java/io/github/s7i/doer/pipeline/Pipeline.java
new file mode 100644
index 0000000..c955d9a
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/Pipeline.java
@@ -0,0 +1,80 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.Globals;
+import io.github.s7i.doer.pipeline.grcp.GrpcInboundConnection;
+import io.github.s7i.doer.pipeline.grcp.GrpcOutboundConnection;
+import io.github.s7i.doer.util.Mark;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import static java.util.Objects.nonNull;
+
+@Slf4j
+public class Pipeline {
+
+    @Mark.Param
+    public static final String DOER_PIPELINE = "doer.pipeline";
+    public static final String DOER_PIPELINE_BACKEND = DOER_PIPELINE + ".backend";
+    public static final String DOER_PIPELINE_BACKEND_TARGET = DOER_PIPELINE_BACKEND + ".target";
+    public static final String DOER_PIPELINE_SINK = DOER_PIPELINE + ".sink";
+    public static final String GRPC_BACKEND = "grpc";
+
+    public static void initFrom(Supplier<Map<String, String>> params) {
+        var setup = params.get().entrySet()
+                .stream()
+                .filter(es -> es.getKey().startsWith(DOER_PIPELINE) && nonNull(es.getValue()))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+        if (!setup.isEmpty()) {
+            Globals.INSTANCE.getPipeline().init(setup);
+        }
+    }
+
+    enum PipelineKind {
+        GRPC_OUTBOUND, GRPC_INBOUND
+    }
+
+    private final Map<PipelineKind, BackendFactory> register = new EnumMap<>(PipelineKind.class);
+
+    void init(Map<String, String> params) {
+        var backend = params.get(DOER_PIPELINE_BACKEND);
+        if (GRPC_BACKEND.equalsIgnoreCase(backend)) {
+            var kind = Boolean.parseBoolean(params.get(DOER_PIPELINE_SINK))
+                    ? PipelineKind.GRPC_INBOUND
+                    : PipelineKind.GRPC_OUTBOUND;
+
+            log.debug("kind: {}", kind);
+
+            switch (kind) {
+                case GRPC_OUTBOUND:
+                    register.put(kind, () -> {
+                        var c = new GrpcOutboundConnection(params.get(DOER_PIPELINE_BACKEND_TARGET));
+                        Globals.INSTANCE.addStopHook(c::closeSafe);
+                        return c;
+                    });
+                    break;
+                case GRPC_INBOUND:
+                    register.put(kind, () -> {
+                        var c = new GrpcInboundConnection(params.get(DOER_PIPELINE_BACKEND_TARGET));
+                        c.setParams(params);
+                        c.connect();
+                        Globals.INSTANCE.addStopHook(c::closeSafe);
+                        return c;
+                    });
+            }
+        }
+    }
+
+    public PipeConnection connect() {
+        return register.values().stream().findFirst().orElseThrow().create();
+    }
+
+    public boolean isEnabled() {
+        return !register.isEmpty();
+    }
+
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/PipelineService.java b/src/main/java/io/github/s7i/doer/pipeline/PipelineService.java
index 4877068..f021f53 100644
--- a/src/main/java/io/github/s7i/doer/pipeline/PipelineService.java
+++ b/src/main/java/io/github/s7i/doer/pipeline/PipelineService.java
@@ -1,72 +1,202 @@
 package io.github.s7i.doer.pipeline;
 
-import io.github.s7i.doer.pipeline.proto.PipelinePublishRequest;
-import io.github.s7i.doer.pipeline.proto.PipelinePublishResponse;
-import io.github.s7i.doer.pipeline.proto.PipelineServiceGrpc;
-import io.grpc.Grpc;
-import io.grpc.InsecureServerCredentials;
-import io.grpc.Server;
-import io.grpc.protobuf.services.HealthStatusManager;
-import io.grpc.protobuf.services.ProtoReflectionService;
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.StringValue;
+import io.github.s7i.doer.domain.grpc.GrpcServer;
+import io.github.s7i.doer.pipeline.proto.*;
+import io.github.s7i.doer.pipeline.store.PipelineStorage;
+import io.github.s7i.doer.pipeline.store.PipelineStorage.Direction;
+import io.github.s7i.doer.pipeline.store.PipelineStorage.Element;
+import io.github.s7i.doer.proto.Record;
+import io.github.s7i.doer.util.Banner;
 import io.grpc.stub.StreamObserver;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 
+import java.io.IOException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
-@Command(name = "pipeline")
+import static java.util.Objects.nonNull;
+
+@Command(name = "pipeline", description = "Pipeline Backend Service.")
 @Slf4j(topic = "doer.console")
-public class PipelineService implements Runnable {
+public class PipelineService implements Callable<Integer>, Banner {
 
-    @Option(names ="--port", defaultValue = "6565")
+    @Option(names = "--port", defaultValue = "6565")
     private Integer port;
-    private Server server;
 
     public static void main(String[] args) {
-        new CommandLine(PipelineService.class).execute(args);
+        System.exit(new CommandLine(PipelineService.class).execute(args));
     }
 
 
+    @Data
+    @RequiredArgsConstructor
+    static class ClientMetadata {
+        final String cid;
+        Element<PipelineLoad> last;
+
+        Element<PipelineLoad> update(Element<PipelineLoad> newElement) {
+            return last = newElement;
+        }
+
+        boolean hasLast() {
+            return nonNull(last);
+        }
+    }
+
     static class Handler extends PipelineServiceGrpc.PipelineServiceImplBase {
+
+        PipelineStorage<PipelineLoad> storage = new PipelineStorage<>();
+        Map<String, ClientMetadata> clientMetadata = new ConcurrentHashMap<>();
+
+
         @Override
         public void publish(PipelinePublishRequest request, StreamObserver<PipelinePublishResponse> responseObserver) {
             var pipelineLoad = request.getPipelineLoad();
             log.info("getting load: {}", pipelineLoad);
 
+            storage.addToPipe("default", pipelineLoad);
+
 
             responseObserver.onNext(PipelinePublishResponse.newBuilder()
                     .setStatus("ok")
                     .build());
             responseObserver.onCompleted();
         }
-    }
 
+        @Override
+        public void exchangeMeta(MetaOp request, StreamObserver<MetaOp> responseObserver) {
 
-    @SneakyThrows
-    @Override
-    public void run() {
-        var hsm = new HealthStatusManager();
+            log.info("new connection {}", request);
+
+            var response = MetaOp.newBuilder();
+
+            if (request.getRequest().getName().equals(Protocol.OP_ADD_NEW)) {
+
+                var resp = MetaOp.Response.newBuilder().setStatus(UUID.randomUUID().toString());
+
+
+                response.setResponse(resp);
+                log.info("response: {}", resp);
+            }
 
-        server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
-                .addService(new Handler())
-                .addService(ProtoReflectionService.newInstance())
-                .addService(hsm.getHealthService())
-                .build()
-                .start();
+            responseObserver.onNext(response.build());
+            responseObserver.onCompleted();
+        }
+
+        @Override
+        public void subscribe(MetaOp request, StreamObserver<PipelineLoad> responseObserver) {
+            log.info("new connection: {}", request);
+
+            var clientMeta = clientMetadata.computeIfAbsent(request.getRequest().getName(), ClientMetadata::new);
+
+            boolean streaming = true;
+            do {
+                storage.getPipe("default").ifPresent(pipe -> {
+                    var es = clientMeta.hasLast()
+                            ? Direction.FIFO.streamFrom(clientMeta.getLast()).skip(1)
+                            : pipe.stream(Direction.FIFO);
+
+                    es.map(clientMeta::update)
+                            .map(Element::getPackage)
+                            .peek(p -> log.info("streaming {}", p))
+                            .forEach(responseObserver::onNext);
+                });
+
+                try {
+                    TimeUnit.MILLISECONDS.sleep(500);
+                } catch (InterruptedException e) {
+                    streaming = false;
+                }
+            } while (streaming);
+
+            responseObserver.onCompleted();
+
+            log.info("stream done");
+        }
+    }
+
+    /**
+     * <pre>
+     * # proto-doer/src/main/proto/doer.proto:14
+     *
+     * message Record {
+     *   google.protobuf.StringValue resource = 1;
+     *   google.protobuf.StringValue key = 2;
+     *   map<string, string> meta = 3;
+     *   google.protobuf.BytesValue data = 4;
+     * }
+     * </pre>
+     */
+    @Command(name = "record", description = "Emit protobuf Record entity.")
+    public void record(
+            @Option(names = "--resource")
+            String resource,
+            @Option(names = "--key")
+            String key,
+            @Option(names = "--meta")
+            Map<String, String> meta,
+            @Option(names = "--data")
+            String data,
+            @Option(names = {"-pl", "--pipeline-load"}, description = "Embed Record in PipelineLoad::Load.")
+            boolean pipelineLoad,
+            @Option(names = {"-h", "--help"}, usageHelp = true)
+            boolean help) {
+
+        var b = Record.newBuilder();
+
+        if (nonNull(resource)) {
+            b.setResource(StringValue.of(resource));
+        }
 
-        log.info("Server Started: {}", server);
+        if (nonNull(key)) {
+            b.setKey(StringValue.of(key));
+        }
 
-        Runtime.getRuntime().addShutdownHook(new Thread(this::stopServer, "grpc-server-shutdown-hook"));
+        if (nonNull(meta)) {
+            b.putAllMeta(meta);
+        }
+
+        if (nonNull(data)) {
+            b.setData(BytesValue.of(ByteString.copyFromUtf8(data)));
+        }
 
-        server.awaitTermination();
-        log.info("bye bye");
+        try {
+            byte[] bytes = pipelineLoad
+                    ? PipelineLoad.newBuilder()
+                    .setLoad(Any.pack(b.build()))
+                    .build()
+                    .toByteArray()
+                    : b.build().toByteArray();
+
+            System.out.write(bytes);
+        } catch (IOException e) {
+            log.error("new record", e);
+        }
     }
 
+
     @SneakyThrows
-    private void stopServer() {
-        server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
+    @Override
+    public Integer call() {
+        printBanner();
+
+        var service = new Handler();
+        new GrpcServer(port, service)
+                .startServer();
+
+        return 0;
     }
 }
diff --git a/src/main/java/io/github/s7i/doer/pipeline/Protocol.java b/src/main/java/io/github/s7i/doer/pipeline/Protocol.java
new file mode 100644
index 0000000..3be2680
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/Protocol.java
@@ -0,0 +1,12 @@
+package io.github.s7i.doer.pipeline;
+
+import io.github.s7i.doer.pipeline.proto.MetaOp;
+
+public class Protocol {
+
+    public interface Operation {
+        MetaOp perform(MetaOp action);
+    }
+
+    public static final String OP_ADD_NEW = "add-new";
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcConnection.java b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcConnection.java
new file mode 100644
index 0000000..5502223
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcConnection.java
@@ -0,0 +1,93 @@
+package io.github.s7i.doer.pipeline.grcp;
+
+import io.github.s7i.doer.DoerException;
+import io.github.s7i.doer.pipeline.PipeConnection;
+import io.github.s7i.doer.pipeline.Protocol;
+import io.github.s7i.doer.pipeline.proto.MetaOp;
+import io.github.s7i.doer.pipeline.proto.PipelineServiceGrpc;
+import io.grpc.Grpc;
+import io.grpc.InsecureChannelCredentials;
+import io.grpc.ManagedChannel;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.util.Objects.requireNonNull;
+
+@RequiredArgsConstructor
+@Slf4j
+public abstract class GrpcConnection implements PipeConnection, AutoCloseable {
+
+    public static final int TIMEOUT = 10;
+    protected ManagedChannel channel;
+    protected PipelineServiceGrpc.PipelineServiceFutureStub serviceFuture;
+    protected PipelineServiceGrpc.PipelineServiceStub serviceStub;
+
+    protected String uuid;
+
+    final String target;
+
+    public void connect() {
+        channel = Grpc.newChannelBuilder(
+                requireNonNull(target, "target"),
+                InsecureChannelCredentials.create()
+        ).build();
+
+        serviceFuture = PipelineServiceGrpc.newFutureStub(channel);
+        serviceStub = PipelineServiceGrpc.newStub(channel);
+
+        int retryNo=0;
+        final int ofRetries=10;
+
+        boolean success = false;
+        do {
+            try {
+                success = introduceNewConnection();
+
+            } catch (RuntimeException | TimeoutException | InterruptedException | ExecutionException e) {
+                log.warn("oops but still working... {}/{}", retryNo + 1, ofRetries, e);
+                try {
+                    int t = 1000 + (1000 * retryNo);
+                    TimeUnit.MILLISECONDS.sleep(t);
+                } catch (InterruptedException x) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        } while (!success && retryNo++ < ofRetries );
+
+        if (!success) {
+            throw new DoerException("[PIPELINE GRPC] FAILURE");
+        }
+        onConnection();
+    }
+
+    protected void onConnection() {
+        log.info("[PIPELINE GRPC] connected");
+    }
+
+    boolean introduceNewConnection() throws ExecutionException, InterruptedException, TimeoutException {
+        uuid = serviceFuture.exchangeMeta(metaNewConnection())
+                .get(TIMEOUT, TimeUnit.SECONDS)
+                .getResponse()
+                .getStatus();
+        log.info("pipeline client id: {}", uuid);
+        return true;
+    }
+
+
+    static MetaOp metaNewConnection() {
+        return MetaOp.newBuilder()
+                .setRequest(MetaOp.Request.newBuilder().setName(Protocol.OP_ADD_NEW))
+                .build();
+    }
+
+    @SneakyThrows
+    public void closeSafe() {
+        close();
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcInboundConnection.java b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcInboundConnection.java
new file mode 100644
index 0000000..713bbec
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcInboundConnection.java
@@ -0,0 +1,79 @@
+package io.github.s7i.doer.pipeline.grcp;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.github.s7i.doer.domain.Mappers;
+import io.github.s7i.doer.pipeline.BlockingPipe;
+import io.github.s7i.doer.pipeline.PipePuller;
+import io.github.s7i.doer.pipeline.proto.MetaOp;
+import io.github.s7i.doer.pipeline.proto.PipelineLoad;
+import io.grpc.stub.StreamObserver;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+@Slf4j
+public class GrpcInboundConnection extends GrpcConnection {
+
+    final BlockingPipe pipe = new BlockingPipe();
+    @Setter
+    Map<String, String> params = Collections.emptyMap();
+
+    public GrpcInboundConnection(String target) {
+        super(requireNonNull(target, "target"));
+    }
+
+    @Override
+    public PipePuller lookupPuller() {
+        return pipe;
+    }
+
+    @Override
+    protected void onConnection() {
+        super.onConnection();
+
+        var meta = MetaOp.newBuilder()
+                .setRequest(MetaOp.Request.newBuilder()
+                        .setName(uuid)
+                        .putAllParameters(params))
+                .build();
+        serviceStub.subscribe(meta, new StreamObserver<>() {
+            @Override
+            public void onNext(PipelineLoad value) {
+                unload(value);
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                log.error("subscribe", t);
+            }
+
+            @Override
+            public void onCompleted() {
+                log.debug("subscribe done");
+            }
+        });
+    }
+
+    private void unload(PipelineLoad value) {
+        try {
+           log.debug("unpack {}", value);
+
+            if (value.hasLoad()) {
+                var load = Mappers.mapFrom(value);
+                pipe.offer(load);
+            }
+        } catch (InvalidProtocolBufferException e) {
+            log.error("oops", e);
+        }
+    }
+
+
+    @Override
+    public void close() throws Exception {
+
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcOutboundConnection.java b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcOutboundConnection.java
new file mode 100644
index 0000000..70f6e5d
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/grcp/GrpcOutboundConnection.java
@@ -0,0 +1,85 @@
+package io.github.s7i.doer.pipeline.grcp;
+
+import com.google.protobuf.Any;
+import io.github.s7i.doer.domain.output.Output;
+import io.github.s7i.doer.pipeline.PipePuller;
+import io.github.s7i.doer.pipeline.proto.PipelineLoad;
+import io.github.s7i.doer.pipeline.proto.PipelinePublishRequest;
+import io.github.s7i.doer.pipeline.proto.PipelinePublishResponse;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.util.Objects.nonNull;
+import static java.util.Objects.requireNonNull;
+
+@Slf4j
+public class GrpcOutboundConnection extends GrpcConnection {
+
+    public GrpcOutboundConnection(String target) {
+        super(target);
+    }
+
+    private Thread thread;
+    private transient boolean isClosed;
+
+    @Override
+    public void registerPuller(PipePuller puller) {
+        connect();
+
+        start(() -> {
+            try {
+                while (!Thread.currentThread().isInterrupted() && !isClosed) {
+                    var load = puller.onNextLoad();
+                    if (nonNull(load)) {
+                        sendLoad(load).ifPresentOrElse(
+                                r -> puller.onAccept(),
+                                () -> log.error("!!! BUG !!! can't happened")
+                        );
+                    }
+                }
+            } catch (Exception e) {
+                log.error("oops", e);
+            }
+            log.debug("[PIPELINE GRPC] : pipeline puller - end of work");
+        });
+    }
+
+    Optional<PipelinePublishResponse> sendLoad(Output.Load load) {
+        requireNonNull(load, "load");
+        try {
+            var request = PipelinePublishRequest.newBuilder()
+                    .setPipelineLoad(PipelineLoad.newBuilder()
+                            .putHeaders("client.uuid", uuid)
+                            .setLoad(Any.pack(load.toRecord())))
+                    .build();
+            return Optional.ofNullable(serviceFuture.publish(request).get(TIMEOUT, TimeUnit.SECONDS));
+        } catch (RuntimeException | TimeoutException | InterruptedException | ExecutionException e) {
+            log.error("oops", e);
+        }
+        return Optional.empty();
+    }
+
+    private void start(Runnable task) {
+        thread = new Thread(task, "grpc-pipeline-puller");
+        //thread.setDaemon(true);
+        thread.start();
+    }
+
+    @Override
+    public void close() throws Exception {
+        log.debug("[PIPELINE GRPC] : closing");
+
+        TimeUnit.MILLISECONDS.sleep(1000);
+
+        isClosed = true;
+
+        if (nonNull(thread)) {
+            thread.interrupt();
+        }
+        channel.shutdownNow().awaitTermination(30, TimeUnit.SECONDS);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/store/ElementIterator.java b/src/main/java/io/github/s7i/doer/pipeline/store/ElementIterator.java
new file mode 100644
index 0000000..d64f80c
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/store/ElementIterator.java
@@ -0,0 +1,72 @@
+package io.github.s7i.doer.pipeline.store;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import static java.util.Objects.*;
+
+public interface ElementIterator<P> {
+
+    class PipeIterator<P> implements Iterator<PipelineStorage.Element<P>> {
+
+        PipelineStorage.Element<P> current;
+        PipelineStorage.Element<P> initial;
+        final PipelineStorage.Direction direction;
+
+        public PipeIterator(PipelineStorage.Direction direction, PipelineStorage.Element<P> initial) {
+            requireNonNull(initial, "initial");
+            requireNonNull(direction, "direction");
+
+            this.direction = direction;
+            this.initial = initial;
+        }
+
+        @Override
+        public boolean hasNext() {
+            if (nonNull(initial)) {
+                return true;
+            }
+
+            return nonNull(current) && direction.from(current).isPresent();
+        }
+
+        @Override
+        public PipelineStorage.Element<P> next() {
+            if (nonNull(initial)) {
+
+                current = initial;
+                initial = null;
+
+                return current;
+            }
+
+            var nextStep = direction.from(requireNonNull(current,
+                    "missing iterator element"
+            ));
+            current = nextStep.get();
+            return current;
+        }
+    }
+
+    PipelineStorage.Element<P> getStartElement(PipelineStorage.Direction direction);
+
+    default Iterable<PipelineStorage.Element<P>> iteratorOf(PipelineStorage.Direction direction, PipelineStorage.Element<P> start) {
+
+        if (isNull(start)) {
+            return Collections::emptyIterator;
+        }
+
+        Supplier<Iterator<PipelineStorage.Element<P>>> prov  = () ->
+                new PipeIterator<>(direction, start);
+
+        return prov::get;
+    }
+
+    default Stream<PipelineStorage.Element<P>> stream(PipelineStorage.Direction direction) {
+        var i = iteratorOf(direction, getStartElement(direction));
+        return StreamSupport.stream(i.spliterator(), false);
+    }
+}
diff --git a/src/main/java/io/github/s7i/doer/pipeline/store/PipelineStorage.java b/src/main/java/io/github/s7i/doer/pipeline/store/PipelineStorage.java
new file mode 100644
index 0000000..1f9d4a3
--- /dev/null
+++ b/src/main/java/io/github/s7i/doer/pipeline/store/PipelineStorage.java
@@ -0,0 +1,125 @@
+package io.github.s7i.doer.pipeline.store;
+
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+
+import java.util.*;
+import java.util.stream.Stream;
+
+import static java.util.Objects.*;
+
+public class PipelineStorage<P> {
+
+    public enum Direction {
+        FIFO, LIFO;
+
+        <T> Optional<Element<T>> from(Element<T> e) {
+            requireNonNull(e, "element");
+            switch (this) {
+                case FIFO:
+                    return e.getPrev();
+                case LIFO:
+                    return e.getNext();
+            }
+            throw new IllegalStateException();
+        }
+
+        public <T> Stream<Element<T>> streamFrom(Element<T> e) {
+            ElementIterator<T> ei = direction -> e;
+            return ei.stream(this);
+        }
+    }
+
+    public interface OffsetSequence {
+        Long nextOffset();
+    }
+
+
+    @Data
+    @Accessors(fluent = true)
+    public static class Pipe<P> implements ElementIterator<P> {
+        Element<P> tail;
+        Element<P> head;
+
+        public void putElementWithPackage(P pack, OffsetSequence oseq) {
+            var link = new Link<P>();
+            var el = new Element<P>(link, oseq.nextOffset(), pack);
+
+            if (nonNull(tail)) {
+                tail.link.prev = el;
+                link.next = tail;
+            }
+            tail = el;
+
+            if (isNull(head)) {
+                head = el;
+            }
+        }
+
+        @Override
+        public Element<P> getStartElement(Direction direction) {
+            return direction == Direction.FIFO ? head : tail;
+        }
+    }
+
+    @RequiredArgsConstructor
+    @ToString
+    public static class Element<P> {
+
+        final Link<P> link;
+
+        final Long offset;
+        final P pack;
+
+        public P getPackage() {
+            return pack;
+        }
+
+        public Optional<Element<P>> getPrev() {
+            return Optional.ofNullable(link.prev);
+        }
+
+        public Optional<Element<P>> getNext() {
+            return Optional.ofNullable(link.next);
+        }
+    }
+
+
+    static class Link<P> {
+        Element<P> prev;
+        Element<P> next;
+
+        @Override
+        public String toString() {
+            return String.format("Link[ prev: %d, next: %d ]",
+                    nonNull(prev) ? prev.offset : -1,
+                    nonNull(next) ? next.offset : -1
+            );
+        }
+    }
+
+    private Map<String, Pipe<P>> store = new HashMap<>();
+    private Long offsetSequence = 0L;
+
+    private Long nextOffset() {
+        return offsetSequence++;
+    }
+
+    public Pipe<P> addToPipe(String pipeName, Collection<P> packages) {
+        var pipe = store.computeIfAbsent(pipeName, name -> new Pipe<>());
+        for (var pack : packages) {
+            pipe.putElementWithPackage(pack, this::nextOffset);
+        }
+        return pipe;
+    }
+
+    public Pipe<P> addToPipe(String pipeName, P pack) {
+        return addToPipe(pipeName, List.of(pack));
+    }
+
+    public Optional<Pipe<P>> getPipe(String name) {
+        return Optional.ofNullable(store.get(name));
+    }
+}
diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml
index 36553fb..e86f79d 100644
--- a/src/main/resources/logback.xml
+++ b/src/main/resources/logback.xml
@@ -23,6 +23,7 @@
     </logger>
 
     <logger name="io.github.s7i.doer" level="DEBUG"/>
+    <logger name="io.grpc" level="INFO"/>
     <logger name="org.apache.helix" level="WARN"/>
     <logger name="org.apache.kafka" level="INFO"/>
     <logger name="org.apache.zookeeper" level="WARN"/>
diff --git a/src/test/groovy/io/github/s7i/doer/command/SinkCommandTest.groovy b/src/test/groovy/io/github/s7i/doer/command/SinkCommandTest.groovy
new file mode 100644
index 0000000..718522e
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/command/SinkCommandTest.groovy
@@ -0,0 +1,16 @@
+package io.github.s7i.doer.command
+
+import spock.lang.Specification
+
+class SinkCommandTest extends Specification {
+
+
+    def "parse manifest test"() {
+        given:
+        def manifestPath = "src/test/resources/pipeline/sink.yml"
+        def sink = ManifestFileCommand.Builder.fromManifestFile(SinkCommand.class, new File(manifestPath))
+
+        expect:
+        sink.onExecuteCommand()
+    }
+}
diff --git a/src/test/groovy/io/github/s7i/doer/command/dump/KafkaDumpTest.groovy b/src/test/groovy/io/github/s7i/doer/command/dump/KafkaDumpTest.groovy
index 7f11f04..761b473 100644
--- a/src/test/groovy/io/github/s7i/doer/command/dump/KafkaDumpTest.groovy
+++ b/src/test/groovy/io/github/s7i/doer/command/dump/KafkaDumpTest.groovy
@@ -15,6 +15,7 @@ import spock.lang.Specification
 
 import java.time.LocalDateTime
 import java.time.ZoneOffset
+import java.util.concurrent.Future
 
 class KafkaDumpTest extends Specification {
 
@@ -144,7 +145,7 @@ class KafkaDumpTest extends Specification {
 
         def prodFactory = Mock(KafkaProducerFactory) {
             1 * createProducer(_, _) >> Mock(Producer) {
-                10 * send(_)
+                10 * send(_, _) >> Mock(Future)
             }
         }
 
diff --git a/src/test/groovy/io/github/s7i/doer/domain/MappersTest.groovy b/src/test/groovy/io/github/s7i/doer/domain/MappersTest.groovy
new file mode 100644
index 0000000..ff8e9a8
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/domain/MappersTest.groovy
@@ -0,0 +1,31 @@
+package io.github.s7i.doer.domain
+
+import com.google.protobuf.Any
+import com.google.protobuf.ByteString
+import com.google.protobuf.BytesValue
+import com.google.protobuf.StringValue
+import io.github.s7i.doer.pipeline.proto.PipelineLoad
+import io.github.s7i.doer.proto.Record
+import spock.lang.Specification
+
+class MappersTest extends Specification {
+    def "MapFrom PipelineLoad into Load"() {
+
+        given:
+        def pipelineLoad = PipelineLoad.newBuilder()
+                .setLoad(Any.pack(Record.newBuilder()
+                        .setData(BytesValue.of(ByteString.copyFromUtf8("this-is-my-string")))
+                        .setKey(StringValue.of("key-123"))
+                        .setResource(StringValue.of("my-res"))
+                        .putMeta("meta-key", "meta-value")
+                        .build()))
+                .build()
+        when:
+        def load = Mappers.mapFrom(pipelineLoad)
+        then:
+        load.dataAsString() == "this-is-my-string"
+        load.getKey() == "key-123"
+        load.getResource() == "my-res"
+        load.getMeta().get("meta-key") == "meta-value"
+    }
+}
diff --git a/src/test/groovy/io/github/s7i/doer/domain/output/PipelineOutputTest.groovy b/src/test/groovy/io/github/s7i/doer/domain/output/PipelineOutputTest.groovy
new file mode 100644
index 0000000..a89632d
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/domain/output/PipelineOutputTest.groovy
@@ -0,0 +1,104 @@
+package io.github.s7i.doer.domain.output
+
+import io.github.s7i.doer.pipeline.BlockingPipe
+import io.github.s7i.doer.pipeline.PipeConnection
+import spock.lang.Specification
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+class PipelineOutputTest extends Specification {
+
+
+    def "Pipeline output test"() {
+        given:
+        def thrSrv = Executors.newFixedThreadPool(1)
+        def load = Output.Load.builder()
+                .data("test".getBytes())
+                .build()
+        BlockingPipe puller
+        def pipeConnection = Mock(PipeConnection) {
+            registerPuller(_) >> { args ->
+                puller = args[0]
+            }
+        }
+        def out = new PipelineOutput(pipeConnection)
+
+        when:
+        out.open()
+        thrSrv.submit({ out.emit(load) })
+        def nextLoad = puller.onNextLoad()
+        puller.onAccept()
+
+        then:
+        thrSrv.shutdown()
+        thrSrv.awaitTermination(1, TimeUnit.SECONDS)
+
+        nextLoad == load
+    }
+
+    def "Pipeline output test with backlog"() {
+        given:
+        def loadTransfer =[]
+        def thrCount=0
+        def thrSrv = Executors.newFixedThreadPool(10, { r->
+            def t = new Thread(r , "pusher-${++thrCount}")
+            t.setDaemon(true)
+            return t
+        })
+        BlockingPipe puller
+        def pipeConnection = Mock(PipeConnection) {
+            registerPuller(_) >> { args ->
+                puller = args[0]
+            }
+        }
+        def out = new PipelineOutput(pipeConnection)
+
+        when:
+        out.open()
+
+        def cdlPush = new CountDownLatch(10)
+
+        10.times {
+            def task = {
+                println "Running task ${Thread.currentThread().getName()}"
+
+                out.emit(Output.Load.builder()
+                    .resource(Thread.currentThread().getName())
+                    .data("test".getBytes())
+                    .build())
+
+                cdlPush.countDown()
+            }
+            thrSrv.submit(task)
+            println "Task submitted (# ${it})"
+        }
+
+        def cdlWorkDone = new CountDownLatch(10)
+        new Thread({
+
+            10.times {
+                def nextLoad = puller.onNextLoad()
+                puller.onAccept()
+
+                loadTransfer << nextLoad
+                cdlWorkDone.countDown()
+
+                println "pull done for ${nextLoad.getResource()}"
+
+            }
+        }, "puller").start()
+
+
+        then:
+        cdlPush.await(5, TimeUnit.SECONDS)
+        cdlWorkDone.await(5, TimeUnit.SECONDS)
+
+        loadTransfer.size() == 10
+
+        thrSrv.shutdown()
+        thrSrv.awaitTermination(5, TimeUnit.SECONDS)
+    }
+
+}
diff --git a/src/test/groovy/io/github/s7i/doer/ingest/record/IngestRecordTest.groovy b/src/test/groovy/io/github/s7i/doer/ingest/record/IngestRecordTest.groovy
new file mode 100644
index 0000000..825eca1
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/ingest/record/IngestRecordTest.groovy
@@ -0,0 +1,39 @@
+package io.github.s7i.doer.ingest.record
+
+import io.github.s7i.doer.Context
+import io.github.s7i.doer.command.YamlParser
+import io.github.s7i.doer.domain.ingest.IngestProcessor
+import io.github.s7i.doer.domain.output.Output
+import io.github.s7i.doer.manifest.ingest.IngestRecordManifest
+import io.github.s7i.doer.util.PropertyResolver
+import spock.lang.Specification
+
+import java.nio.file.Path
+
+class IngestRecordTest extends Specification {
+
+    def "Parse Manifest"() {
+
+        given:
+        YamlParser yml = { Path.of("src/test/resources/ingest/ingest-record.yml").toFile() }
+        def manifest = yml.parseYaml(IngestRecordManifest.class)
+
+        def records = []
+        def context = Mock(Context) {
+            getParams() >> ["doer.output": "mock-me"]
+            getPropertyResolver() >> new PropertyResolver()
+            buildOutput(_) >> Mock(Output) {
+                3 * emit(_) >> {
+                    records << it[0]
+                }
+            }
+        }
+
+        new IngestProcessor(context).process(manifest)
+
+        expect:
+        manifest.getKind() == 'ingest'
+        records.size() == 3
+        (records[2] as Output.Load).getKey() == "key-record 3"
+    }
+}
diff --git a/src/test/groovy/io/github/s7i/doer/pipeline/PipelineTest.groovy b/src/test/groovy/io/github/s7i/doer/pipeline/PipelineTest.groovy
new file mode 100644
index 0000000..1308f16
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/pipeline/PipelineTest.groovy
@@ -0,0 +1,29 @@
+package io.github.s7i.doer.pipeline
+
+import io.github.s7i.doer.Globals
+import io.github.s7i.doer.domain.ConfigProcessor
+import spock.lang.Specification
+
+import java.nio.file.Path
+
+class PipelineTest extends Specification {
+
+    def "should init pipeline"() {
+        given:
+        def params = [
+                "doer.pipeline.backend"                 : "kafka",
+                "doer.pipeline.backend.kafka.properties": "bootstrap.servers=kafka:9093\n"
+
+        ]
+        def pipeline = Mock(Pipeline) {
+        }
+        Globals.INSTANCE.pipeline = pipeline
+
+        when:
+        new ConfigProcessor(Path.of("src/test/resources/pipeline/pipeline-config.yml").toFile()).processConfig()
+
+        then:
+        1 * pipeline.init(params)
+
+    }
+}
diff --git a/src/test/groovy/io/github/s7i/doer/pipeline/store/PipelineStorageTest.groovy b/src/test/groovy/io/github/s7i/doer/pipeline/store/PipelineStorageTest.groovy
new file mode 100644
index 0000000..fa79f9d
--- /dev/null
+++ b/src/test/groovy/io/github/s7i/doer/pipeline/store/PipelineStorageTest.groovy
@@ -0,0 +1,66 @@
+package io.github.s7i.doer.pipeline.store
+
+import spock.lang.Specification
+
+import java.util.stream.Collectors
+
+class PipelineStorageTest extends Specification {
+    def "AddToPipe - one element"() {
+        given:
+        def ps = new PipelineStorage<String>()
+        when:
+        def pipe = ps.addToPipe("pipe-name", "pack-of-data")
+
+        then:
+        pipe
+
+        pipe.head.offset == 0
+        pipe.tail.getPackage() == "pack-of-data"
+    }
+
+
+    def "AddToPipe - 5 elements"() {
+        given:
+        def ps = new PipelineStorage<String>()
+        def els = []
+        5.times {
+            els << "package-of-data-${it + 1}"
+        }
+
+        when:
+        def pipe = ps.addToPipe("pipe-name", els)
+
+        then:
+        pipe
+
+        pipe.tail.getPackage() == "package-of-data-5"
+        pipe.head.getPackage() == "package-of-data-1"
+
+        pipe.tail.offset == 4L
+        pipe.head.offset == 0L
+    }
+
+    def "Pipe - iterate"() {
+        given:
+        def ps = new PipelineStorage<String>()
+        def els = []
+        5.times {
+            els << "data-${it + 1}"
+        }
+
+        when:
+        def pipe = ps.addToPipe("pipe-name", els)
+        def collected = pipe.stream(direction)
+                .map { it.getPackage() }
+                .collect(Collectors.toList())
+
+        then:
+        collected == expected
+
+        where:
+        direction                      | expected
+        PipelineStorage.Direction.FIFO | ["data-1", "data-2", "data-3", "data-4", "data-5"]
+        PipelineStorage.Direction.LIFO | ["data-5", "data-4", "data-3", "data-2", "data-1"]
+    }
+
+}
diff --git a/src/test/resources/ingest/ingest-record.yml b/src/test/resources/ingest/ingest-record.yml
new file mode 100644
index 0000000..ccca702
--- /dev/null
+++ b/src/test/resources/ingest/ingest-record.yml
@@ -0,0 +1,7 @@
+version: v1
+kind: ingest
+ingest:
+  - record: record 1 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+  - record: record 2 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+  - record: record 3 ${date:yyyy-MM-dd}T${date:HH:mm:ss}+00:00
+    key: key-record 3
diff --git a/src/test/resources/pipeline/ingest.yml b/src/test/resources/pipeline/ingest.yml
new file mode 100644
index 0000000..723488e
--- /dev/null
+++ b/src/test/resources/pipeline/ingest.yml
@@ -0,0 +1,7 @@
+version: v1
+kind: ingest
+params:
+  doer.pipeline.bind: from-ingest-to-sink
+  doer.pipeline.id: id-ingest1
+spec:
+  - record: my first record
diff --git a/src/test/resources/pipeline/pipeline-config.yml b/src/test/resources/pipeline/pipeline-config.yml
new file mode 100644
index 0000000..5d41b73
--- /dev/null
+++ b/src/test/resources/pipeline/pipeline-config.yml
@@ -0,0 +1,7 @@
+version: v1
+kind: Config
+params:
+  doer.pipeline.backend: kafka
+  doer.pipeline.backend.kafka.properties: |+
+    bootstrap.servers=kafka:9093
+  noop.param: random
diff --git a/src/test/resources/pipeline/sink.yml b/src/test/resources/pipeline/sink.yml
new file mode 100644
index 0000000..a31fe85
--- /dev/null
+++ b/src/test/resources/pipeline/sink.yml
@@ -0,0 +1,7 @@
+version: v1
+kind: sink
+params:
+  doer.pipeline.bind: from-ingest-to-sink
+  doer.pipeline.id: id-sink1
+spec:
+  - output: doer://console