Repository: hyperledger/quilt
Pull request #253 (2019-09-14 00:06:31)

Add Link modules (STREAM-PR3)
* Add link-core contains all interfaces and abstractions for links.
* Add link-ilp-over-http contains an implementation of ILP-over-HTTP using OkHttp3.
* Unit tests.
* Fix Javadoc in various places.

Signed-off-by: David Fuelling <sappenin@gmail.com>

codecov[bot]: # [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=h1) Report
> Merging [#253](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=desc) into [df/stream-pr2-stream-codecs](https://codecov.io/gh/hyperledger/quilt/commit/45fa582766a276ebd4b35933104ffcd42587a597?src=pr&el=desc) will **decrease** coverage by `2.91%`.
> The diff coverage is `55.59%`.

[![Impacted file tree graph](https://codecov.io/gh/hyperledger/quilt/pull/253/graphs/tree.svg?width=650&token=XEfCq6zC6W&height=150&src=pr)](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=tree)

```diff
@@                        Coverage Diff                        @@
##             df/stream-pr2-stream-codecs     #253      +/-   ##
=================================================================
- Coverage                          69.79%   66.87%   -2.92%     
- Complexity                           480      553      +73     
=================================================================
  Files                                174      203      +29     
  Lines                               2251     2832     +581     
  Branches                             164      182      +18     
=================================================================
+ Hits                                1571     1894     +323     
- Misses                               606      855     +249     
- Partials                              74       83       +9
```


| [Impacted Files](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...rg/interledger/stream/frames/StreamCloseFrame.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY29yZS9zcmMvbWFpbi9qYXZhL29yZy9pbnRlcmxlZGdlci9zdHJlYW0vZnJhbWVzL1N0cmVhbUNsb3NlRnJhbWUuamF2YQ==) | `0% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: |
| [...rledger/link/exceptions/LinkNotFoundException.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvZXhjZXB0aW9ucy9MaW5rTm90Rm91bmRFeGNlcHRpb24uamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [.../interledger/link/http/IlpOverHttpLinkFactory.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1pbHAtb3Zlci1odHRwL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvaHR0cC9JbHBPdmVySHR0cExpbmtGYWN0b3J5LmphdmE=) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...rc/main/java/org/interledger/link/Connectable.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvQ29ubmVjdGFibGUuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...rc/main/java/org/interledger/link/LinkFactory.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvTGlua0ZhY3RvcnkuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...rg/interledger/link/http/IlpOverHttpConstants.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1pbHAtb3Zlci1odHRwL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvaHR0cC9JbHBPdmVySHR0cENvbnN0YW50cy5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...dger/link/http/auth/SimpleBearerTokenSupplier.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1pbHAtb3Zlci1odHRwL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvaHR0cC9hdXRoL1NpbXBsZUJlYXJlclRva2VuU3VwcGxpZXIuamF2YQ==) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...er/link/http/auth/JwtHs256BearerTokenSupplier.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1pbHAtb3Zlci1odHRwL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvaHR0cC9hdXRoL0p3dEhzMjU2QmVhcmVyVG9rZW5TdXBwbGllci5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...nk/events/AbstractLinkConnectionEventListener.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvZXZlbnRzL0Fic3RyYWN0TGlua0Nvbm5lY3Rpb25FdmVudExpc3RlbmVyLmphdmE=) | `0% <0%> (ø)` | `0 <0> (?)` | |
| [...ger/link/exceptions/LinkNotConnectedException.java](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvZXhjZXB0aW9ucy9MaW5rTm90Q29ubmVjdGVkRXhjZXB0aW9uLmphdmE=) | `0% <0%> (ø)` | `0 <0> (?)` | |
| ... and [49 more](https://codecov.io/gh/hyperledger/quilt/pull/253/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=footer). Last update [45fa582...672c2f0](https://codecov.io/gh/hyperledger/quilt/pull/253?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

sappenin: This PR depends on #244 and #252. Those should be merged to `master` first, and then we can change the target of this PR.

Patch:
diff --git a/link-parent/link-core/pom.xml b/link-parent/link-core/pom.xml
new file mode 100644
index 0000000..be4e157
--- /dev/null
+++ b/link-parent/link-core/pom.xml
@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <groupId>org.interledger</groupId>
+    <artifactId>link-parent</artifactId>
+    <version>0.18.0-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <name>Quilt :: Link :: Core</name>
+  <artifactId>link-core</artifactId>
+  <description>A Link abstraction for connecting Interledger peers.</description>
+  <packaging>jar</packaging>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>ilp-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.immutables</groupId>
+      <artifactId>value</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+    </dependency>
+  </dependencies>
+</project>
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/AbstractLink.java b/link-parent/link-core/src/main/java/org/interledger/link/AbstractLink.java
new file mode 100644
index 0000000..d3a8f82
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/AbstractLink.java
@@ -0,0 +1,99 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.exceptions.LinkHandlerAlreadyRegisteredException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+/**
+ * An abstract implementation of a {@link Link} that provides scaffolding for all link implementations.
+ */
+public abstract class AbstractLink<L extends LinkSettings> implements Link<L> {
+
+  protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  /**
+   * The optionally-present {@link InterledgerAddress} of the operator of this sender. Sometimes the Operator Address is
+   * not yet populated when this client is constructed (e.g, IL-DCP). Thus, the value is optional to accommodate these
+   * cases.
+   */
+  private final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier;
+
+  /**
+   * A typed representation of the configuration options passed-into this ledger link.
+   */
+  private final L linkSettings;
+
+  private final AtomicReference<LinkHandler> linkHandlerAtomicReference = new AtomicReference<>();
+
+  // Non-final for late-binding...
+  private final AtomicReference<LinkId> linkId = new AtomicReference<>();
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} that supplies the optionally-present operator address of the node
+   *                                operating this Link (note this is optional to support IL-DCP).
+   * @param linkSettings            A {@link L} that specified ledger link options.
+   */
+  protected AbstractLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier,
+      final L linkSettings
+  ) {
+    this.operatorAddressSupplier = Objects.requireNonNull(operatorAddressSupplier);
+    this.linkSettings = Objects.requireNonNull(linkSettings);
+  }
+
+  @Override
+  public LinkId getLinkId() {
+    if (linkId.get() == null) {
+      throw new IllegalStateException("The LinkId must be set before using a Link");
+    }
+    return linkId.get();
+  }
+
+  @Override
+  public void setLinkId(final LinkId linkId) {
+    if (!this.linkId.compareAndSet(null, Objects.requireNonNull(linkId))) {
+      throw new IllegalStateException("LinkId may only be set once");
+    }
+  }
+
+  @Override
+  public Supplier<Optional<InterledgerAddress>> getOperatorAddressSupplier() {
+    return operatorAddressSupplier;
+  }
+
+  @Override
+  public L getLinkSettings() {
+    return this.linkSettings;
+  }
+
+  @Override
+  public void registerLinkHandler(final LinkHandler ilpDataHandler)
+      throws LinkHandlerAlreadyRegisteredException {
+    Objects.requireNonNull(ilpDataHandler, "ilpDataHandler must not be null!");
+    if (!this.linkHandlerAtomicReference.compareAndSet(null, ilpDataHandler)) {
+      throw new LinkHandlerAlreadyRegisteredException(
+          "DataHandler may not be registered twice. Call unregisterDataHandler first!",
+          this.getLinkId()
+      );
+    }
+  }
+
+  @Override
+  public void unregisterLinkHandler() {
+    this.linkHandlerAtomicReference.set(null);
+  }
+
+  @Override
+  public Optional<LinkHandler> getLinkHandler() {
+    return Optional.ofNullable(linkHandlerAtomicReference.get());
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/AbstractStatefulLink.java b/link-parent/link-core/src/main/java/org/interledger/link/AbstractStatefulLink.java
new file mode 100644
index 0000000..710a24a
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/AbstractStatefulLink.java
@@ -0,0 +1,231 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.events.LinkConnectedEvent;
+import org.interledger.link.events.LinkConnectionEventEmitter;
+import org.interledger.link.events.LinkConnectionEventListener;
+import org.interledger.link.events.LinkDisconnectedEvent;
+
+import com.google.common.eventbus.EventBus;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+
+/**
+ * An abstract implementation of a {@link Link} that provides scaffolding for all link implementations.
+ */
+public abstract class AbstractStatefulLink<L extends LinkSettings>
+    extends AbstractLink<L> implements StatefulLink<L> {
+
+  private final AtomicBoolean connected = new AtomicBoolean(NOT_CONNECTED);
+
+  // The emitter used by this link.
+  private final LinkConnectionEventEmitter linkConnectionEventEmitter;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param operatorAddressSupplier    A {@link Supplier} for the ILP address of the node operating this link.
+   * @param linkSettings               A {@link L} that specified ledger link options.
+   * @param linkConnectionEventEmitter A {@link LinkConnectionEventEmitter} that is used to emit events from this link.
+   */
+  protected AbstractStatefulLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier,
+      final L linkSettings,
+      final LinkConnectionEventEmitter linkConnectionEventEmitter
+  ) {
+    super(operatorAddressSupplier, linkSettings);
+    this.linkConnectionEventEmitter = Objects.requireNonNull(linkConnectionEventEmitter);
+  }
+
+  @Override
+  public final CompletableFuture<Void> connect() {
+    try {
+      if (this.connected.compareAndSet(NOT_CONNECTED, CONNECTED)) {
+        logger.debug("[linktype={}] (operatorIlpAddress={}) connecting to linkId={}...",
+            this.getLinkSettings().getLinkType(),
+            this.operatorAddressAsString(),
+            this.getLinkId()
+        );
+
+        return this.doConnect()
+            .whenComplete(($, error) -> {
+              if (error == null) {
+                // Emit a connected event...
+                this.linkConnectionEventEmitter.emitEvent(LinkConnectedEvent.of(this));
+
+                logger.debug("[linktype={}] (Operator operatorIlpAddress={}) connected to linkId={}",
+                    this.getLinkSettings().getLinkType(),
+                    this.operatorAddressAsString(),
+                    this.getLinkId()
+                );
+              } else {
+                this.connected.set(NOT_CONNECTED);
+                final String errorMessage = String.format(
+                    "[linktype=%s] (Operator operatorIlpAddress=%s) was unable to connect to linkId=%s",
+                    this.getLinkSettings().getLinkType(),
+                    this.operatorAddressAsString(),
+                    this.getLinkId().value()
+                );
+                logger.error(errorMessage, error);
+              }
+            });
+      } else {
+        logger.debug("[linktype={}] (Operator operatorIlpAddress={})  already connected to linkId={}",
+            this.getLinkSettings().getLinkType(),
+            this.operatorAddressAsString(),
+            this.getLinkId()
+        );
+        // No-op: We're already connected.
+        return CompletableFuture.completedFuture(null);
+      }
+    } catch (RuntimeException e) {
+      // If we can't connect, then disconnect this account in order to trigger any listeners.
+      this.disconnect().join();
+      throw e;
+    } catch (Exception e) {
+      // If we can't connect, then disconnect this account in order to trigger any listeners.
+      this.disconnect().join();
+      throw new RuntimeException(e.getMessage(), e);
+    }
+  }
+
+  /**
+   * Perform the logic of actually connecting to the remote peer.
+   *
+   * @return A {@link CompletableFuture} that completes when the method is completed.
+   */
+  public abstract CompletableFuture<Void> doConnect();
+
+  @Override
+  public void close() {
+    this.disconnect().join();
+  }
+
+  @Override
+  public final CompletableFuture<Void> disconnect() {
+    try {
+      if (this.connected.compareAndSet(CONNECTED, NOT_CONNECTED)) {
+        logger.debug("[linktype={}] (ILPAddress={}) disconnecting from linkId={}...",
+            this.getLinkSettings().getLinkType(),
+            this.operatorAddressAsString(),
+            this.getLinkId()
+        );
+
+        return this.doDisconnect()
+            .whenComplete(($, error) -> {
+              if (error == null) {
+                // emit disconnected event.
+                this.linkConnectionEventEmitter.emitEvent(LinkDisconnectedEvent.of(this));
+
+                logger.debug("[linktype={}] (ILPAddress={}) disconnected from linkId={}",
+                    this.getLinkSettings().getLinkType(),
+                    this.operatorAddressAsString(),
+                    this.getLinkId()
+                );
+              } else {
+                final String errorMessage =
+                    String.format(
+                        "[linktype=%s] While trying to disconnect linkId=%s from operatorIlpAddress=%s error=%s",
+                        this.getLinkSettings().getLinkType(),
+                        this.getLinkId(),
+                        this.operatorAddressAsString(),
+                        error.getMessage()
+                    );
+                logger.error(errorMessage, error);
+              }
+            })
+            .thenAccept(($) -> logger.debug("[linktype={}] (ILPAddress={}) disconnected from linkId={}",
+                this.getLinkSettings().getLinkType(),
+                this.operatorAddressAsString(),
+                this.getLinkId()
+            ));
+      } else {
+        logger.debug("[linktype={}] (ILPAddress={}) already disconnected from linkId={}",
+            this.getLinkSettings().getLinkType(),
+            this.operatorAddressAsString(),
+            this.getLinkId());
+        // No-op: We're already disconnected.
+        return CompletableFuture.completedFuture(null);
+      }
+    } catch (Exception e) {
+      throw new RuntimeException(e.getMessage(), e);
+    }
+  }
+
+  /**
+   * Perform the logic of disconnecting from the remote peer.
+   *
+   * @return A {@link CompletableFuture} that completes when the method is completed.
+   */
+  public abstract CompletableFuture<Void> doDisconnect();
+
+  @Override
+  public boolean isConnected() {
+    return this.connected.get();
+  }
+
+  @Override
+  public void addLinkEventListener(final LinkConnectionEventListener linkConnectionEventListener) {
+    Objects.requireNonNull(linkConnectionEventListener);
+    this.linkConnectionEventEmitter.addLinkConnectionEventListener(linkConnectionEventListener);
+  }
+
+  @Override
+  public void removeLinkEventListener(final LinkConnectionEventListener linkConnectionEventListener) {
+    Objects.requireNonNull(linkConnectionEventListener);
+    this.linkConnectionEventEmitter.removeLinkConnectionEventListener(linkConnectionEventListener);
+  }
+
+  /**
+   * A helper method to properly obtain this Node's ILP operating address as a {@link String}, or a consistent value if
+   * the address has not yet been set.
+   *
+   * @return A {@link String}.
+   */
+  private String operatorAddressAsString() {
+    return this.getOperatorAddressSupplier().get().map(InterledgerAddress::getValue).orElse("unset");
+  }
+
+
+  /**
+   * An example {@link LinkConnectionEventEmitter} that allows Link-related events to be emitted into an EventBus.
+   */
+  public static class EventBusConnectionEventEmitter implements LinkConnectionEventEmitter {
+
+    private final EventBus eventBus;
+
+    public EventBusConnectionEventEmitter(final EventBus eventBus) {
+      this.eventBus = Objects.requireNonNull(eventBus);
+    }
+
+    /////////////////
+    // Event Emitters
+    /////////////////
+
+    @Override
+    public void emitEvent(final LinkConnectedEvent event) {
+      eventBus.post(event);
+    }
+
+    @Override
+    public void emitEvent(final LinkDisconnectedEvent event) {
+      eventBus.post(event);
+    }
+
+    @Override
+    public void addLinkConnectionEventListener(final LinkConnectionEventListener linkConnectionEventListener) {
+      Objects.requireNonNull(linkConnectionEventListener);
+      eventBus.register(linkConnectionEventListener);
+    }
+
+    @Override
+    public void removeLinkConnectionEventListener(final LinkConnectionEventListener linkConnectionEventListener) {
+      Objects.requireNonNull(linkConnectionEventListener);
+      eventBus.unregister(linkConnectionEventListener);
+    }
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/Connectable.java b/link-parent/link-core/src/main/java/org/interledger/link/Connectable.java
new file mode 100644
index 0000000..2a062d6
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/Connectable.java
@@ -0,0 +1,40 @@
+package org.interledger.link;
+
+import java.io.Closeable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Defines how to connect and disconnect.
+ */
+public interface Connectable extends Closeable {
+
+  boolean CONNECTED = true;
+  boolean NOT_CONNECTED = false;
+
+  /**
+   * <p>Connect to the remote peer.</p>
+   *
+   * @return A {@link CompletableFuture} with the result of this call.
+   */
+  CompletableFuture<Void> connect();
+
+  /**
+   * Disconnect from the remote peer.
+   *
+   * @return A {@link CompletableFuture} with the result of this call.
+   */
+  CompletableFuture<Void> disconnect();
+
+  @Override
+  default void close() {
+    disconnect().join();
+  }
+
+  /**
+   * <p>Determines if this Connectable is connected or not. If authentication is required, this method
+   * will return false until an authenticated session is opened.</p>
+   *
+   * @return {@code true} if this Connectable is connected; {@code false} otherwise.
+   */
+  boolean isConnected();
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/Ids.java b/link-parent/link-core/src/main/java/org/interledger/link/Ids.java
new file mode 100644
index 0000000..caebef7
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/Ids.java
@@ -0,0 +1,38 @@
+package org.interledger.link;
+
+import org.interledger.core.Wrapped;
+import org.interledger.core.Wrapper;
+
+import org.immutables.value.Value;
+
+/**
+ * Wrapped immutable classes for providing type-safe identifiers.
+ */
+public class Ids {
+
+  @Value.Immutable
+  @Wrapped
+  static abstract class _LinkId extends Wrapper<String> {
+  }
+
+  @Value.Immutable
+  @Wrapped
+  static abstract class _LinkType extends Wrapper<String> {
+
+    /**
+     * Always normalize Link-type String values to full uppercase to avoid casing ambiguity in properties files.
+     *
+     * @return A Link type.
+     */
+    @Value.Check
+    public _LinkType normalize() {
+      final String linkTypeString = this.value();
+      if (!linkTypeString.toUpperCase().equals(linkTypeString)) {
+        return LinkType.of(linkTypeString.toUpperCase());
+      } else {
+        return this;
+      }
+    }
+
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/Link.java b/link-parent/link-core/src/main/java/org/interledger/link/Link.java
new file mode 100644
index 0000000..655d7b9
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/Link.java
@@ -0,0 +1,107 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerResponsePacket;
+import org.interledger.link.exceptions.LinkHandlerAlreadyRegisteredException;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * <p>An abstraction for communicating with a remote Interledger peer in a connection-state-less manner, meaning links
+ * do not hold connection state during their lifecyle.</p>
+ *
+ * <p>If you want to make a {@link Link} that can holds connection state (such as for a link protocol using
+ * Websockets), consider {@link StatefulLink} instead.</p>
+ */
+public interface Link<L extends LinkSettings> extends LinkSender {
+
+  /**
+   * <p>A unique identifier for this {@link Link} so that the same link can be referenced across requests.</p>
+   *
+   * <p>Note that this value is non-optional because even though an account operating on a Link may lazily load its
+   * Interledger address, a Link will always have an Id because it always corresponds to a single account, which always
+   * has an accountId.</p>
+   *
+   * @return A {@link LinkId}.
+   */
+  LinkId getLinkId();
+
+  /**
+   * Allows the linkId to be set after construction. This is important because often times a LinkId is not known until
+   * after it is constructed (i.e., it cannot be discerned from {@link LinkSettings}).
+   *
+   * @param linkId The {@link LinkId} to set on this Link.
+   */
+  void setLinkId(LinkId linkId);
+
+  /**
+   * A supplier for the ILP address of this node operator. This value may be empty, for example, in cases where the
+   * Link obtains its address from a parent node using IL-DCP.
+   *
+   * @return A {@link Supplier} of an optioanally-present {@link InterledgerAddress}.
+   */
+  Supplier<Optional<InterledgerAddress>> getOperatorAddressSupplier();
+
+  /**
+   * The settings for this Link.
+   *
+   * @return An instance of {@link L}.
+   */
+  L getLinkSettings();
+
+  /**
+   * <p>Set the callback which is used to handle incoming prepared data packets. The handler should expect one
+   * parameter (an ILP Prepare Packet) and return a {@link InterledgerResponsePacket} as a response. If an error occurs,
+   * the callback MAY throw an exception. In general, however, the callback should behave as {@link
+   * Link#sendPacket(InterledgerPreparePacket)} does.</p>
+   *
+   * <p>If a data handler is already set, this method throws a {@link LinkHandlerAlreadyRegisteredException}. In order
+   * to change the data handler, the old handler must first be removed via {@link #unregisterLinkHandler()}. This is to
+   * ensure that handlers are not overwritten by accident.</p>
+   *
+   * <p>If an incoming packet is received by the link, but no handler is registered, the link SHOULD respond with
+   * an error.</p>
+   *
+   * @param dataHandler An instance of {@link LinkHandler}.
+   */
+  void registerLinkHandler(LinkHandler dataHandler) throws LinkHandlerAlreadyRegisteredException;
+
+  /**
+   * Accessor for the currently registered (though optionally-present) {@link LinkHandler}.
+   *
+   * @return An optionally-present {@link LinkHandler}.
+   */
+  Optional<LinkHandler> getLinkHandler();
+
+  /**
+   * Accessor for the currently registered (though optionally-present) {@link LinkHandler}.
+   *
+   * @return The currently registered {@link LinkHandler}.
+   *
+   * @throws RuntimeException if no handler is registered (A Link is not in a valid state until it has handlers
+   *                          registered)
+   */
+  default LinkHandler safeGetLinkHandler() {
+    return this.getLinkHandler()
+        .orElseThrow(() -> new IllegalStateException("You MUST register a LinkHandler before using this link!"));
+  }
+
+  /**
+   * Removes the currently used {@link LinkHandler}. This has the same effect as if {@link
+   * #registerLinkHandler(LinkHandler)} had never been called. If no link handler is currently set, this method does
+   * nothing.
+   */
+  void unregisterLinkHandler();
+
+  /**
+   * <p>Check the connection's connectivity. to see  for ping by making an HTTP Head request with a ping packet, and
+   * asserting the values returned are one of the supported content-types required for BLAST.</p>
+   *
+   * <p>If the endpoint does not support producing BLAST responses, we expect a 406 NOT_ACCEPTABLE response. If the
+   * endpoint does not support BLAST requests, then we expect a 415 UNSUPPORTED_MEDIA_TYPE.</p>
+   */
+  default void testConnection() {
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LinkFactory.java b/link-parent/link-core/src/main/java/org/interledger/link/LinkFactory.java
new file mode 100644
index 0000000..36fdadc
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LinkFactory.java
@@ -0,0 +1,60 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * A factory for constructing instances of {@link Link} based upon configured settings.
+ */
+public interface LinkFactory {
+
+  /**
+   * Construct a new instance of {@link Link} using the supplied inputs.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} that supplies the ILP address of the Connector operating this
+   *                                link.
+   * @param linkSettings            An instance of {@link LinkSettings} to initialize this link from.
+   *
+   * @return A newly constructed instance of {@link Link}.
+   */
+  Link<?> constructLink(Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, LinkSettings linkSettings);
+
+  /**
+   * Helper method to apply custom settings on a per-link-type basis.
+   *
+   * @param linkSettings A {@link LinkSettings} to apply to this instance.
+   *
+   * @return A {@link LinkSettings} with applied custom settings.
+   */
+  default LinkSettings applyCustomSettings(LinkSettings linkSettings) {
+    return linkSettings;
+  }
+
+  /**
+   * Determines if this factory support a particular type of {@link LinkType}.
+   *
+   * @param linkType A {@link LinkType} to check compatibility for.
+   *
+   * @return {@code true} if this factory supports the specified linkType; {@code false} otherwise.
+   */
+  boolean supports(LinkType linkType);
+
+  /**
+   * Construct a new instance of {@link Link} using the supplied inputs.
+   *
+   * @param $                       A {@link Class} to satisfy Java generics.
+   * @param operatorAddressSupplier A {@link Supplier} for the address of the node operating this factory.
+   * @param linkSettings            A {@link LinkSettings} to use in order to construct a {@link Link}.
+   * @param <LS>                    A type that extends {@link LinkSettings}.
+   * @param <L>                     A type that extends {@link Link}.
+   *
+   * @return An instance of {@link L}.
+   */
+  default <LS extends LinkSettings, L extends Link<LS>> L constructLink(
+      final Class<L> $, Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, final LS linkSettings
+  ) {
+    return (L) this.constructLink(operatorAddressSupplier, linkSettings);
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LinkFactoryProvider.java b/link-parent/link-core/src/main/java/org/interledger/link/LinkFactoryProvider.java
new file mode 100644
index 0000000..af2db1e
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LinkFactoryProvider.java
@@ -0,0 +1,42 @@
+package org.interledger.link;
+
+import org.interledger.link.exceptions.LinkException;
+
+import com.google.common.collect.Maps;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * A provider of {@link LinkFactory} scoped by type.
+ */
+public class LinkFactoryProvider {
+
+  // At runtime, there will potentially be _many_ factories depending on the LinkType (e.g., a BTP Factory, a
+  // LoopbackFactory, etc).
+  private final Map<LinkType, LinkFactory> linkFactories;
+
+  public LinkFactoryProvider() {
+    this(Maps.newConcurrentMap());
+  }
+
+  public LinkFactoryProvider(final Map<LinkType, LinkFactory> linkFactories) {
+    this.linkFactories = Objects.requireNonNull(linkFactories);
+  }
+
+  public LinkFactory getLinkFactory(final LinkType linkType) {
+    Objects.requireNonNull(linkType, "linkType must not be null");
+    return Optional.ofNullable(this.linkFactories.get(linkType))
+        .orElseThrow(() -> new LinkException(
+            String.format("No registered LinkFactory linkType=%s", linkType), LinkId.of("n/a"))
+        );
+  }
+
+  public LinkFactory registerLinkFactory(final LinkType linkType, final LinkFactory linkFactory) {
+    Objects.requireNonNull(linkType, "linkType must not be null");
+    Objects.requireNonNull(linkFactory, "linkFactory must not be null");
+
+    return this.linkFactories.put(linkType, linkFactory);
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LinkHandler.java b/link-parent/link-core/src/main/java/org/interledger/link/LinkHandler.java
new file mode 100644
index 0000000..e1f4df6
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LinkHandler.java
@@ -0,0 +1,26 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerFulfillPacket;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Handles an incoming {@link InterledgerPreparePacket} for a single plugin, sent from a remote peer.
+ */
+@FunctionalInterface
+public interface LinkHandler {
+
+  /**
+   * Handles an incoming {@link InterledgerPreparePacket} received from a connected peer, but that may have originated
+   * from any Interledger sender in the network.
+   *
+   * @param incomingPreparePacket A {@link InterledgerPreparePacket} containing data about an incoming payment.
+   *
+   * @return A {@link CompletableFuture} that resolves to an optionally-present {@link InterledgerResponsePacket}, which
+   * will be of concrete type {@link InterledgerFulfillPacket} or {@link InterledgerRejectPacket}, if present.
+   */
+  InterledgerResponsePacket handleIncomingPacket(InterledgerPreparePacket incomingPreparePacket);
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LinkSender.java b/link-parent/link-core/src/main/java/org/interledger/link/LinkSender.java
new file mode 100644
index 0000000..f9a0ce5
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LinkSender.java
@@ -0,0 +1,40 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerFulfillPacket;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * Defines how to send data to the other side of a bilateral link (i.e., the other party operating a single account in
+ * tandem with the operator of this sender).
+ */
+@FunctionalInterface
+public interface LinkSender {
+
+  /**
+   * <p>Sends an ILPv4 request packet to a peer and returns the response packet (if one is returned).</p>
+   *
+   * <p>This method supports one of three responses, which can be handled by using either
+   * {@link InterledgerResponsePacket#handle(Consumer, Consumer)} or {@link InterledgerResponsePacket#map(Function,
+   * Function)}.
+   * </p>
+   *
+   * <ul>
+   *   <li>An instance of {@link InterledgerFulfillPacket}, which means the packet was fulfilled by the receiver.</li>
+   *   <li>An instance of {@link InterledgerRejectPacket}, which means the packet was rejected by one of the nodes in
+   *   the payment path.
+   *   </li>
+   * </ul>
+   *
+   * @param preparePacket An {@link InterledgerPreparePacket} to send to the remote peer.
+   *
+   * @return An {@link InterledgerResponsePacket}, which will be of concrete type {@link InterledgerFulfillPacket} or }
+   *     } or {@link InterledgerRejectPacket}.
+   */
+  InterledgerResponsePacket sendPacket(InterledgerPreparePacket preparePacket);
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LinkSettings.java b/link-parent/link-core/src/main/java/org/interledger/link/LinkSettings.java
new file mode 100644
index 0000000..3588b3e
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LinkSettings.java
@@ -0,0 +1,60 @@
+package org.interledger.link;
+
+import org.immutables.value.Value;
+
+import java.util.Map;
+
+/**
+ * Configuration information relating to a {@link Link}.
+ */
+public interface LinkSettings {
+
+  static ImmutableLinkSettings.Builder builder() {
+    return ImmutableLinkSettings.builder();
+  }
+
+  /**
+   * The type of this ledger link.
+   *
+   * @return A {@link LinkType}
+   */
+  LinkType getLinkType();
+
+  /**
+   * Additional, custom settings that any link can define.
+   *
+   * @return A {@link Map} with {@link String} keys.
+   */
+  Map<String, Object> getCustomSettings();
+
+  @Value.Immutable
+  abstract class AbstractLinkSettings implements LinkSettings {
+
+    /**
+     * Additional, custom settings that any link can define. Redacted to prevent credential leakage in log files.
+     *
+     * @return A {@link Map} with {@link String} keys.
+     */
+    @Value.Redacted
+    public abstract Map<String, Object> getCustomSettings();
+
+    /**
+     * Always normalize Link-type String values to full uppercase to avoid casing ambiguity in properties files.
+     *
+     * @return A {@link AbstractLinkSettings} with normalized values.
+     */
+    @Value.Check
+    public AbstractLinkSettings normalize() {
+      final String linkTypeString = this.getLinkType().value();
+      if (!linkTypeString.toUpperCase().equals(linkTypeString)) {
+        return ImmutableLinkSettings.builder()
+            .from(this)
+            .linkType(LinkType.of(linkTypeString.toUpperCase()))
+            .build();
+      } else {
+        return this;
+      }
+    }
+
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLink.java b/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLink.java
new file mode 100644
index 0000000..db72030
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLink.java
@@ -0,0 +1,87 @@
+package org.interledger.link;
+
+import static org.interledger.core.InterledgerConstants.ALL_ZEROS_FULFILLMENT;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerFulfillPacket;
+import org.interledger.core.InterledgerFulfillment;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerResponsePacket;
+import org.interledger.link.exceptions.LinkHandlerAlreadyRegisteredException;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * <p>A {@link Link} that always responds with a Fulfillment that contains the data supplied by the Prepare packet.</p>
+ */
+public class LoopbackLink extends AbstractLink<LinkSettings> implements Link<LinkSettings> {
+
+  public static final String LINK_TYPE_STRING = "LOOPBACK";
+  public static final LinkType LINK_TYPE = LinkType.of(LINK_TYPE_STRING);
+
+  public static final InterledgerFulfillment LOOPBACK_FULFILLMENT = ALL_ZEROS_FULFILLMENT;
+
+  // A constant key that can be added to this link's customSettings to simulate a particular rejection error code.
+  public static final String SIMULATED_REJECT_ERROR_CODE = "simulatedRejectErrorCode";
+
+  private final PacketRejector packetRejector;
+
+  /**
+   * Required-Args Constructor.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} of the ILP address for the node that's operating this Link.
+   * @param linkSettings            A {@link LinkSettings} for this Link.
+   * @param packetRejector          A {@link PacketRejector} to aid in rejecting packets in a uniform manner.
+   */
+  public LoopbackLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier,
+      final LinkSettings linkSettings,
+      final PacketRejector packetRejector
+  ) {
+    super(operatorAddressSupplier, linkSettings);
+    this.packetRejector = Objects.requireNonNull(packetRejector);
+  }
+
+  @Override
+  public void registerLinkHandler(LinkHandler ilpDataHandler) throws LinkHandlerAlreadyRegisteredException {
+    throw new RuntimeException(
+        "Loopback links never have incoming data, and thus should not have a registered DataHandler."
+    );
+  }
+
+  @Override
+  public InterledgerResponsePacket sendPacket(final InterledgerPreparePacket preparePacket) {
+    Objects.requireNonNull(preparePacket, "preparePacket must not be null");
+
+    return Optional.ofNullable(this.getLinkSettings().getCustomSettings().get(SIMULATED_REJECT_ERROR_CODE))
+        .map((value) -> {
+          if (value.equals("T02")) {
+            return packetRejector.reject(this.getLinkId(), preparePacket, InterledgerErrorCode.T02_PEER_BUSY,
+                "Loopback set to manually reject via simulate_timeout=T02");
+          } else if (value.equals("T03")) {
+            // Sleep for 1 minute, which in the typical case will exceed the Circuit-breaker's threshold.
+            try {
+              Thread.sleep(60000);
+            } catch (InterruptedException e) {
+              throw new RuntimeException(e);
+            }
+            return packetRejector.reject(this.getLinkId(), preparePacket, InterledgerErrorCode.T03_CONNECTOR_BUSY,
+                "Loopback set to exceed timeout via simulate_timeout=T03");
+          }
+          if (value.equals("T99")) {
+            throw new RuntimeException("T99 APPLICATION ERROR");
+          } else {
+            return InterledgerFulfillPacket.builder()
+                .fulfillment(LOOPBACK_FULFILLMENT)
+                .data(preparePacket.getData())
+                .build();
+          }
+        })
+        .orElseGet(InterledgerFulfillPacket.builder()
+            .fulfillment(LOOPBACK_FULFILLMENT)
+            .data(preparePacket.getData())::build);
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLinkFactory.java b/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLinkFactory.java
new file mode 100644
index 0000000..c104464
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/LoopbackLinkFactory.java
@@ -0,0 +1,52 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.exceptions.LinkException;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * An implementation of {@link LinkFactory} for creating Links that can handle the `Loopback` packets.
+ */
+public class LoopbackLinkFactory implements LinkFactory {
+
+  private final PacketRejector packetRejector;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param packetRejector An instance of {@link PacketRejector}.
+   */
+  public LoopbackLinkFactory(final PacketRejector packetRejector) {
+    this.packetRejector = Objects.requireNonNull(packetRejector, "packetRejector must not be null");
+  }
+
+  /**
+   * Construct a new instance of {@link Link} using the supplied inputs.
+   *
+   * @return A newly constructed instance of {@link Link}.
+   */
+  public Link<?> constructLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, final LinkSettings linkSettings
+  ) {
+    Objects.requireNonNull(operatorAddressSupplier, "operatorAddressSupplier must not be null");
+    Objects.requireNonNull(linkSettings, "linkSettings must not be null");
+
+    if (!this.supports(linkSettings.getLinkType())) {
+      throw new LinkException(
+          String.format("LinkType not supported by this factory. linkType=%s", linkSettings.getLinkType()),
+          LinkId.of("n/a")
+      );
+    }
+
+    return new LoopbackLink(operatorAddressSupplier, linkSettings, packetRejector);
+  }
+
+  @Override
+  public boolean supports(LinkType linkType) {
+    return LoopbackLink.LINK_TYPE.equals(linkType);
+  }
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/PacketRejector.java b/link-parent/link-core/src/main/java/org/interledger/link/PacketRejector.java
new file mode 100644
index 0000000..38ca5ad
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/PacketRejector.java
@@ -0,0 +1,70 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * Utility class for helping to reject a packet.
+ */
+public class PacketRejector {
+
+  public static final InterledgerAddress UNSET_OPERATOR_ADDRESS =
+      InterledgerAddress.of(InterledgerAddress.AllocationScheme.PRIVATE.getValue() + ".unset-operator-address");
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  private final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} of the ILP address for the node that's operating this Link.
+   */
+  public PacketRejector(final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier) {
+    this.operatorAddressSupplier = Objects
+        .requireNonNull(operatorAddressSupplier, "operatorAddressSupplier must not be null");
+  }
+
+  /**
+   * Helper-method to reject a request.
+   *
+   * @param rejectingLinkId The {@link Link} that was the source of this Reject action.
+   * @param preparePacket   The {@link InterledgerPreparePacket} that was the catalyst for this reject action.
+   * @param errorCode       The {@link InterledgerErrorCode} that expresses the reason for this reject action.
+   * @param errorMessage    An error message for clarity. If no error message is desired, supply an empty string.
+   *
+   * @return An {@link InterledgerRejectPacket} constructed using the above inputs.
+   */
+  public InterledgerRejectPacket reject(
+      final LinkId rejectingLinkId, final InterledgerPreparePacket preparePacket,
+      final InterledgerErrorCode errorCode, final String errorMessage
+  ) {
+    Objects.requireNonNull(rejectingLinkId, "rejectingLinkId must not be null");
+    Objects.requireNonNull(preparePacket, "preparePacket must not be null");
+    Objects.requireNonNull(errorCode, "errorCode must not be null");
+    Objects.requireNonNull(errorMessage, "errorMessage must not be null");
+
+    // Reject.
+    final InterledgerRejectPacket rejectPacket = InterledgerRejectPacket.builder()
+        .triggeredBy(operatorAddressSupplier.get().orElse(UNSET_OPERATOR_ADDRESS))
+        .code(errorCode)
+        .message(errorMessage)
+        .build();
+
+    logger.debug(
+        "Rejecting inside linkId={}: PreparePacket={} RejectPacket={}",
+        rejectingLinkId, preparePacket, rejectPacket
+    );
+
+    return rejectPacket;
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLink.java b/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLink.java
new file mode 100644
index 0000000..9b03b53
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLink.java
@@ -0,0 +1,81 @@
+package org.interledger.link;
+
+import static org.interledger.link.PacketRejector.UNSET_OPERATOR_ADDRESS;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerCondition;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerFulfillPacket;
+import org.interledger.core.InterledgerFulfillment;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+import org.interledger.link.exceptions.LinkHandlerAlreadyRegisteredException;
+
+import java.util.Base64;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * <p>A {@link Link} that responds to Ping protocol packets that conform to the Ping protocol.</p>
+ *
+ * <p>Ping functionality is implemented as a Loopback Link so that all packet-processing related to balance tracking
+ * can be properly performed without any special processing inside of the core connector logic. Additionally, this link
+ * is a loopback link so that no outbound traffic ever leaves the Connector while processing a Ping request.
+ * </p>
+ */
+public class PingLoopbackLink extends AbstractLink<LinkSettings> implements Link<LinkSettings> {
+
+  public static final String LINK_TYPE_STRING = "PING";
+  public static final LinkType LINK_TYPE = LinkType.of(LINK_TYPE_STRING);
+
+  public static final InterledgerFulfillment PING_PROTOCOL_FULFILLMENT =
+      InterledgerFulfillment.of(Base64.getDecoder().decode("cGluZ3BpbmdwaW5ncGluZ3BpbmdwaW5ncGluZ3Bpbmc="));
+  public static final InterledgerCondition PING_PROTOCOL_CONDITION = InterledgerCondition.of(
+      Base64.getDecoder().decode("jAC8DGFPZPfh4AtZpXuvXFe2oRmpDVSvSJg2oT+bx34="));
+
+  /**
+   * Required-Args Constructor.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} of the ILP address for the node that's operating this Link.
+   * @param linkSettings            A {@link LinkSettings} for this Link.
+   */
+  public PingLoopbackLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, final LinkSettings linkSettings
+  ) {
+    super(operatorAddressSupplier, linkSettings);
+  }
+
+  @Override
+  public void registerLinkHandler(LinkHandler ilpDataHandler) throws LinkHandlerAlreadyRegisteredException {
+    throw new RuntimeException(
+        "PingLoopback links never have incoming data, and thus should not have a registered DataHandler."
+    );
+  }
+
+  @Override
+  public InterledgerResponsePacket sendPacket(final InterledgerPreparePacket preparePacket) {
+    Objects.requireNonNull(preparePacket, "preparePacket must not be null!");
+
+    if (preparePacket.getExecutionCondition().equals(PING_PROTOCOL_CONDITION)) {
+      return InterledgerFulfillPacket.builder()
+          .fulfillment(PING_PROTOCOL_FULFILLMENT)
+          .data(preparePacket.getData())
+          .build();
+    } else {
+      // Reject.
+      final InterledgerRejectPacket rejectPacket = InterledgerRejectPacket.builder()
+          .triggeredBy(getOperatorAddressSupplier().get().orElse(UNSET_OPERATOR_ADDRESS))
+          .code(InterledgerErrorCode.F00_BAD_REQUEST)
+          .message("Invalid Ping Protocol Condition")
+          .build();
+
+      logger.warn(
+          "Rejecting Unidirectional Ping packet: PreparePacket={} RejectPacket={}", preparePacket, rejectPacket
+      );
+
+      return rejectPacket;
+    }
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLinkFactory.java b/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLinkFactory.java
new file mode 100644
index 0000000..13e5608
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/PingLoopbackLinkFactory.java
@@ -0,0 +1,53 @@
+package org.interledger.link;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.exceptions.LinkException;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * An implementation of {@link LinkFactory} for creating instances of {@link Link} that can handle `Unidirectional Ping`
+ * packets.
+ */
+public class PingLoopbackLinkFactory implements LinkFactory {
+
+  private PingLoopbackLink lazyPingLoopbackLink;
+
+  /**
+   * Construct a new instance of {@link Link} using the supplied inputs.
+   *
+   * @return A newly constructed instance of {@link Link}.
+   */
+  public Link<?> constructLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, final LinkSettings linkSettings
+  ) {
+    Objects.requireNonNull(operatorAddressSupplier, "operatorAddressSupplier must not be null");
+    Objects.requireNonNull(linkSettings, "linkSettings must not be null");
+
+    if (!this.supports(linkSettings.getLinkType())) {
+      throw new LinkException(
+          String.format("LinkType not supported by this factory. linkType=%s", linkSettings.getLinkType()),
+          LinkId.of("n/a")
+      );
+    }
+
+    if (lazyPingLoopbackLink != null) {
+      return lazyPingLoopbackLink;
+    } else {
+      synchronized (this) {
+        if (lazyPingLoopbackLink == null) {
+          lazyPingLoopbackLink = new PingLoopbackLink(operatorAddressSupplier, linkSettings);
+        }
+        return lazyPingLoopbackLink;
+      }
+    }
+  }
+
+  @Override
+  public boolean supports(LinkType linkType) {
+    return PingLoopbackLink.LINK_TYPE.equals(linkType);
+  }
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/StatefulLink.java b/link-parent/link-core/src/main/java/org/interledger/link/StatefulLink.java
new file mode 100644
index 0000000..a1a7518
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/StatefulLink.java
@@ -0,0 +1,31 @@
+package org.interledger.link;
+
+import org.interledger.link.events.LinkConnectionEventListener;
+
+/**
+ * <p>Defines a {@link Link} that holds connection state and possibly other stateful data. Because this type of {@link
+ * Link} holds a connection, it cannot be load-balance in the typical fashion. For example, a Link with an underlying
+ * Websocket connection cannot easily be transitioned and serviced by another ILP node instance in a different
+ * runtime.</p>
+ */
+public interface StatefulLink<LS extends LinkSettings> extends Link<LS>, Connectable {
+
+  /**
+   * Add an event listener to this link.
+   *
+   * Care should be taken when adding multiple listeners to ensure that they perform distinct operations, otherwise
+   * duplicate functionality might be unintentionally introduced.
+   *
+   * @param eventListener A {@link LinkConnectionEventListener} that can listen for and response to various types of
+   *                      events emitted by this link.
+   */
+  void addLinkEventListener(LinkConnectionEventListener eventListener);
+
+  /**
+   * Removes an event listener from the collection of listeners registered with this link.
+   *
+   * @param eventListener A {@link LinkConnectionEventListener} representing the listener to remove.
+   */
+  void removeLinkEventListener(LinkConnectionEventListener eventListener);
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/AbstractLinkConnectionEventListener.java b/link-parent/link-core/src/main/java/org/interledger/link/events/AbstractLinkConnectionEventListener.java
new file mode 100644
index 0000000..08a9e7e
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/AbstractLinkConnectionEventListener.java
@@ -0,0 +1,18 @@
+package org.interledger.link.events;
+
+/**
+ * An abstract implementation of {@link LinkConnectionEventListener} that no-ops all methods except those relating to
+ * {@link LinkConnectionEventListener}, which are left to the implementation to define.
+ */
+public abstract class AbstractLinkConnectionEventListener implements LinkConnectionEventListener {
+
+  @Override
+  public void onConnect(LinkConnectedEvent event) {
+    // No-op by default.
+  }
+
+  @Override
+  public void onDisconnect(LinkDisconnectedEvent event) {
+    // No-op by default.
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectedEvent.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectedEvent.java
new file mode 100644
index 0000000..8c2c41b
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectedEvent.java
@@ -0,0 +1,22 @@
+package org.interledger.link.events;
+
+import org.interledger.link.Link;
+
+import org.immutables.value.Value;
+
+/**
+ * Emitted after a {@link Link} connects to a remote peer.
+ */
+public interface LinkConnectedEvent extends LinkConnectionEvent {
+
+  static LinkConnectedEvent of(final Link<?> link) {
+    return ImmutableLinkConnectedEvent.builder().link(link).build();
+  }
+
+  @Value.Immutable
+  abstract class AbstractLinkConnectedEvent implements
+      LinkConnectedEvent {
+
+  }
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEvent.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEvent.java
new file mode 100644
index 0000000..f1e2052
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEvent.java
@@ -0,0 +1,8 @@
+package org.interledger.link.events;
+
+/**
+ * Defines functionality for all Link events relating to connections.
+ */
+public interface LinkConnectionEvent extends LinkEvent {
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventEmitter.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventEmitter.java
new file mode 100644
index 0000000..e449aeb
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventEmitter.java
@@ -0,0 +1,41 @@
+package org.interledger.link.events;
+
+import org.interledger.link.Link;
+
+import java.util.UUID;
+
+/**
+ * Defines how a link should emit connection-related events. Note that a given {@link Link} has only a single connection
+ * event-emitter.
+ */
+public interface LinkConnectionEventEmitter {
+
+  /**
+   * Emit an event of type {@link LinkConnectedEvent}.
+   *
+   * @param event The event to emit.
+   */
+  void emitEvent(LinkConnectedEvent event);
+
+  /**
+   * Emit an event of type {@link LinkDisconnectedEvent}.
+   *
+   * @param event The event to emit.
+   */
+  void emitEvent(LinkDisconnectedEvent event);
+
+  /**
+   * Add a DataLink event listener to this emitter.
+   *
+   * @param eventlistener A {@link LinkConnectionEventListener} that can handle various types of events emitted by this
+   *                      ledger link.
+   */
+  void addLinkConnectionEventListener(LinkConnectionEventListener eventlistener);
+
+  /**
+   * Removes an event listener from this emitter.
+   *
+   * @param eventlistener A {@link UUID} representing the unique identifier of the listener, as seen by this emitter.
+   */
+  void removeLinkConnectionEventListener(LinkConnectionEventListener eventlistener);
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventListener.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventListener.java
new file mode 100644
index 0000000..c2954f6
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkConnectionEventListener.java
@@ -0,0 +1,24 @@
+package org.interledger.link.events;
+
+import org.interledger.link.Link;
+
+/**
+ * Handler interface that defines how to listen for events emitted by a {@link Link}.
+ */
+public interface LinkConnectionEventListener {
+
+  /**
+   * Called to handle an {@link LinkConnectedEvent}.
+   *
+   * @param event A {@link LinkConnectedEvent}.
+   */
+  void onConnect(LinkConnectedEvent event);
+
+  /**
+   * Called to handle an {@link LinkDisconnectedEvent}.
+   *
+   * @param event A {@link LinkDisconnectedEvent}.
+   */
+  void onDisconnect(LinkDisconnectedEvent event);
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkDisconnectedEvent.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkDisconnectedEvent.java
new file mode 100644
index 0000000..0338bd1
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkDisconnectedEvent.java
@@ -0,0 +1,22 @@
+package org.interledger.link.events;
+
+import org.interledger.link.Link;
+
+import org.immutables.value.Value;
+
+/**
+ * Emitted after a {@link Link} disconnects from a remote peer.
+ */
+public interface LinkDisconnectedEvent extends LinkConnectionEvent {
+
+  static LinkDisconnectedEvent of(final Link<?> link) {
+    return ImmutableLinkDisconnectedEvent.builder().link(link).build();
+  }
+
+  @Value.Immutable
+  abstract class AbstractLinkDisconnectedEvent implements
+      LinkDisconnectedEvent {
+
+  }
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/events/LinkEvent.java b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkEvent.java
new file mode 100644
index 0000000..7196c8a
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/events/LinkEvent.java
@@ -0,0 +1,32 @@
+package org.interledger.link.events;
+
+import org.interledger.link.Link;
+
+import com.google.common.collect.Maps;
+import org.immutables.value.Value.Default;
+
+import java.util.Map;
+
+/**
+ * A parent interface for all link events.
+ */
+public interface LinkEvent {
+
+  /**
+   * Accessor for the {@link Link} that emitted this event.
+   *
+   * @return The {@link Link} that emitted this event.
+   */
+  Link<?> getLink();
+
+  /**
+   * Custom properties that can be added to any DataLink event.
+   *
+   * @return A {@link Map} of custom settings.
+   */
+  @Default
+  default Map<String, Object> getCustomSettings() {
+    return Maps.newConcurrentMap();
+  }
+
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkException.java b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkException.java
new file mode 100644
index 0000000..bd26de9
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkException.java
@@ -0,0 +1,88 @@
+package org.interledger.link.exceptions;
+
+import org.interledger.link.LinkId;
+
+/**
+ * A root exception for all exceptions relating to ILPv4 Links.
+ */
+public class LinkException extends RuntimeException {
+
+  /**
+   * The account-address of the link that threw this exception.
+   */
+  private final LinkId linkId;
+
+  /**
+   * Constructs a new runtime exception with {@code null} as its detail message.  The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkException(LinkId linkId) {
+    this.linkId = linkId;
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()}
+   *                method.
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkException(String message, LinkId linkId) {
+    super(message);
+    this.linkId = linkId;
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message and cause.  <p>Note that the detail message
+   * associated with {@code cause} is <i>not</i> automatically incorporated in this runtime exception's detail message.
+   *
+   * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+   * @param cause   the cause (which is saved for later retrieval by the {@link #getCause()} method).  (A
+   *                <tt>null</tt> value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkException(String message, Throwable cause, LinkId linkId) {
+    super(message, cause);
+    this.linkId = linkId;
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified cause and a detail message of
+   * <tt>(cause==null ? null : cause.toString())</tt> (which typically contains the class and detail
+   * message of <tt>cause</tt>).  This constructor is useful for runtime exceptions that are little more than wrappers
+   * for other throwables.
+   *
+   * @param cause  the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <tt>null</tt>
+   *               value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkException(Throwable cause, LinkId linkId) {
+    super(cause);
+    this.linkId = linkId;
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and
+   * writable stack trace enabled or disabled.
+   *
+   * @param message            the detail message.
+   * @param cause              the cause.  (A {@code null} value is permitted, and indicates that the cause is
+   *                           nonexistent or unknown.)
+   * @param enableSuppression  whether or not suppression is enabled or disabled
+   * @param writableStackTrace whether or not the stack trace should be writable
+   * @param linkId             The {@link LinkId} that triggered this exception.
+   */
+  public LinkException(
+      String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace, LinkId linkId
+  ) {
+    super(message, cause, enableSuppression, writableStackTrace);
+    this.linkId = linkId;
+  }
+
+  public LinkId getLinkId() {
+    return linkId;
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkHandlerAlreadyRegisteredException.java b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkHandlerAlreadyRegisteredException.java
new file mode 100644
index 0000000..29aa794
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkHandlerAlreadyRegisteredException.java
@@ -0,0 +1,81 @@
+package org.interledger.link.exceptions;
+
+
+import org.interledger.link.Link;
+import org.interledger.link.LinkHandler;
+import org.interledger.link.LinkId;
+
+/**
+ * Thrown if a caller attempts to register a {@link LinkHandler} in a {@link Link} where one is already registered.
+ */
+public class LinkHandlerAlreadyRegisteredException extends LinkException {
+
+  /**
+   * Constructs a new runtime exception with {@code null} as its detail message.  The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkHandlerAlreadyRegisteredException(LinkId linkId) {
+    super(linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()}
+   *                method.
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkHandlerAlreadyRegisteredException(String message, LinkId linkId) {
+    super(message, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message and cause.  <p>Note that the detail message
+   * associated with {@code cause} is <i>not</i> automatically incorporated in this runtime exception's detail message.
+   *
+   * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+   * @param cause   the cause (which is saved for later retrieval by the {@link #getCause()} method).  (A <tt>null</tt>
+   *                value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkHandlerAlreadyRegisteredException(String message, Throwable cause,
+      LinkId linkId) {
+    super(message, cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified cause and a detail message of
+   * <tt>(cause==null ? null : cause.toString())</tt> (which typically contains the class and detail
+   * message of <tt>cause</tt>).  This constructor is useful for runtime exceptions that are little more than wrappers
+   * for other throwables.
+   *
+   * @param cause  the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <tt>null</tt>
+   *               value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkHandlerAlreadyRegisteredException(Throwable cause,
+      LinkId linkId) {
+    super(cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and
+   * writable stack trace enabled or disabled.
+   *
+   * @param message            the detail message.
+   * @param cause              the cause.  (A {@code null} value is permitted, and indicates that the cause is
+   *                           nonexistent or unknown.)
+   * @param enableSuppression  whether or not suppression is enabled or disabled
+   * @param writableStackTrace whether or not the stack trace should be writable
+   * @param linkId             The {@link LinkId} that triggered this exception.
+   */
+  public LinkHandlerAlreadyRegisteredException(String message, Throwable cause,
+      boolean enableSuppression,
+      boolean writableStackTrace,
+      LinkId linkId) {
+    super(message, cause, enableSuppression, writableStackTrace, linkId);
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotConnectedException.java b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotConnectedException.java
new file mode 100644
index 0000000..2d82a56
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotConnectedException.java
@@ -0,0 +1,74 @@
+package org.interledger.link.exceptions;
+
+import org.interledger.link.LinkId;
+
+/**
+ * Thrown if an operation is attempted on an un-connected ledger link.
+ */
+public class LinkNotConnectedException extends LinkException {
+
+  /**
+   * Constructs a new runtime exception with {@code null} as its detail message.  The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotConnectedException(LinkId linkId) {
+    super(linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()}
+   *                method.
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotConnectedException(String message, LinkId linkId) {
+    super(message, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message and cause.  <p>Note that the detail message
+   * associated with {@code cause} is <i>not</i> automatically incorporated in this runtime exception's detail message.
+   *
+   * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+   * @param cause   the cause (which is saved for later retrieval by the {@link #getCause()} method).  (A <tt>null</tt>
+   *                value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotConnectedException(String message, Throwable cause, LinkId linkId) {
+    super(message, cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified cause and a detail message of
+   * <tt>(cause==null ? null : cause.toString())</tt> (which typically contains the class and detail
+   * message of <tt>cause</tt>).  This constructor is useful for runtime exceptions that are little more than wrappers
+   * for other throwables.
+   *
+   * @param cause  the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <tt>null</tt>
+   *               value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotConnectedException(Throwable cause, LinkId linkId) {
+    super(cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and
+   * writable stack trace enabled or disabled.
+   *
+   * @param message            the detail message.
+   * @param cause              the cause.  (A {@code null} value is permitted, and indicates that the cause is
+   *                           nonexistent or unknown.)
+   * @param enableSuppression  whether or not suppression is enabled or disabled
+   * @param writableStackTrace whether or not the stack trace should be writable
+   * @param linkId             The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotConnectedException(String message, Throwable cause, boolean enableSuppression,
+      boolean writableStackTrace, LinkId linkId) {
+    super(message, cause, enableSuppression, writableStackTrace, linkId);
+  }
+}
diff --git a/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotFoundException.java b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotFoundException.java
new file mode 100644
index 0000000..e8821a4
--- /dev/null
+++ b/link-parent/link-core/src/main/java/org/interledger/link/exceptions/LinkNotFoundException.java
@@ -0,0 +1,74 @@
+package org.interledger.link.exceptions;
+
+import org.interledger.link.LinkId;
+
+/**
+ * Thrown if an operation is attempted on an un-connected ledger link.
+ */
+public class LinkNotFoundException extends LinkException {
+
+  /**
+   * Constructs a new runtime exception with {@code null} as its detail message.  The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotFoundException(LinkId linkId) {
+    super(linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may
+   * subsequently be initialized by a call to {@link #initCause}.
+   *
+   * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()}
+   *                method.
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotFoundException(String message, LinkId linkId) {
+    super(message, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message and cause.  <p>Note that the detail message
+   * associated with {@code cause} is <i>not</i> automatically incorporated in this runtime exception's detail message.
+   *
+   * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+   * @param cause   the cause (which is saved for later retrieval by the {@link #getCause()} method).  (A <tt>null</tt>
+   *                value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId  The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotFoundException(String message, Throwable cause, LinkId linkId) {
+    super(message, cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified cause and a detail message of
+   * <tt>(cause==null ? null : cause.toString())</tt> (which typically contains the class and detail
+   * message of <tt>cause</tt>).  This constructor is useful for runtime exceptions that are little more than wrappers
+   * for other throwables.
+   *
+   * @param cause  the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <tt>null</tt>
+   *               value is permitted, and indicates that the cause is nonexistent or unknown.)
+   * @param linkId The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotFoundException(Throwable cause, LinkId linkId) {
+    super(cause, linkId);
+  }
+
+  /**
+   * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and
+   * writable stack trace enabled or disabled.
+   *
+   * @param message            the detail message.
+   * @param cause              the cause.  (A {@code null} value is permitted, and indicates that the cause is
+   *                           nonexistent or unknown.)
+   * @param enableSuppression  whether or not suppression is enabled or disabled
+   * @param writableStackTrace whether or not the stack trace should be writable
+   * @param linkId             The {@link LinkId} that triggered this exception.
+   */
+  public LinkNotFoundException(String message, Throwable cause, boolean enableSuppression,
+      boolean writableStackTrace, LinkId linkId) {
+    super(message, cause, enableSuppression, writableStackTrace, linkId);
+  }
+}
diff --git a/link-parent/link-ilp-over-http/pom.xml b/link-parent/link-ilp-over-http/pom.xml
new file mode 100644
index 0000000..14d1afe
--- /dev/null
+++ b/link-parent/link-ilp-over-http/pom.xml
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <groupId>org.interledger</groupId>
+    <artifactId>link-parent</artifactId>
+    <version>0.18.0-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <name>Quilt :: Link :: ILP-over-HTTP</name>
+  <artifactId>link-ilp-over-http</artifactId>
+  <description>A Link implementation for ILP-over-HTTP</description>
+  <packaging>jar</packaging>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>codecs-framework</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>ilp-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>link-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.auth0</groupId>
+      <artifactId>java-jwt</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.squareup.okhttp3</groupId>
+      <artifactId>okhttp</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.immutables</groupId>
+      <artifactId>value</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.zalando</groupId>
+      <artifactId>problem</artifactId>
+    </dependency>
+
+  </dependencies>
+</project>
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpConstants.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpConstants.java
new file mode 100644
index 0000000..f93ed8b
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpConstants.java
@@ -0,0 +1,17 @@
+package org.interledger.link.http;
+
+import static com.google.common.net.MediaType.OCTET_STREAM;
+
+import okhttp3.MediaType;
+
+/**
+ * Common headers used in the BLAST protocol.
+ */
+public interface IlpOverHttpConstants {
+
+  String OCTET_STREAM_STRING = OCTET_STREAM.toString();
+  MediaType APPLICATION_OCTET_STREAM = MediaType.parse(OCTET_STREAM_STRING);
+
+  String BEARER = "Bearer ";
+  String ILP_OPERATOR_ADDRESS_VALUE = "ILP-Operator-Address";
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java
new file mode 100644
index 0000000..6769118
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java
@@ -0,0 +1,298 @@
+package org.interledger.link.http;
+
+import static com.google.common.net.HttpHeaders.CACHE_CONTROL;
+import static com.google.common.net.HttpHeaders.CONTENT_TYPE;
+import static com.google.common.net.HttpHeaders.PRAGMA;
+import static com.google.common.net.MediaType.OCTET_STREAM;
+import static org.interledger.link.http.IlpOverHttpConstants.APPLICATION_OCTET_STREAM;
+import static org.interledger.link.http.IlpOverHttpConstants.BEARER;
+import static org.interledger.link.http.IlpOverHttpConstants.OCTET_STREAM_STRING;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerConstants;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+import org.interledger.encoding.asn.framework.CodecContext;
+import org.interledger.link.AbstractLink;
+import org.interledger.link.Link;
+import org.interledger.link.LinkHandler;
+import org.interledger.link.LinkType;
+import org.interledger.link.exceptions.LinkException;
+import org.interledger.link.http.auth.BearerTokenSupplier;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.io.CharStreams;
+import com.google.common.net.HttpHeaders;
+import okhttp3.Headers;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Request.Builder;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.zalando.problem.Problem;
+import org.zalando.problem.ThrowableProblem;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * <p>An extension of {@link AbstractLink} that handles HTTP (aka, ILP over HTTP) connections, both incoming and
+ * outgoing.</p>
+ *
+ * <p>To handle incoming HTTP requests, use {@link #registerLinkHandler(LinkHandler)}.</p>
+ *
+ * @see "https://github.com/interledger/rfcs/blob/master/0035-ilp-over-http/0035-ilp-over-http.md"
+ */
+public class IlpOverHttpLink extends AbstractLink<IlpOverHttpLinkSettings> implements Link<IlpOverHttpLinkSettings> {
+
+  public static final String LINK_TYPE_STRING = "ILP_OVER_HTTP";
+  public static final LinkType LINK_TYPE = LinkType.of(LINK_TYPE_STRING);
+
+  private static final InterledgerPreparePacket UNFULFILLABLE_PACKET = InterledgerPreparePacket.builder()
+      .executionCondition(InterledgerConstants.ALL_ZEROS_CONDITION)
+      .expiresAt(Instant.now().plusSeconds(30))
+      .destination(InterledgerAddress.of("peer.ilp_over_http_connection_test_that_should_always_reject"))
+      .build();
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  // Note: The Http client in this sender is shared between all HTTP links...
+  private final OkHttpClient okHttpClient;
+  private final ObjectMapper objectMapper;
+  private final CodecContext ilpCodecContext;
+
+  /**
+   * These are fine to be typed as String for two reasons: 1) They will be put into headers as a String (so they'll hang
+   * around in memory) and 2) they have when they're JWTs, they have short expiries (the signing key is not persisted as
+   * a String).
+   */
+  private final Supplier<String> authTokenSupplier;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param operatorAddressSupplier A {@link Supplier} of an optionally present {@link InterledgerAddress} for the
+   *                                operator of this link.
+   * @param ilpOverHttpLinkSettings A {@link IlpOverHttpLinkSettings} that specified ledger link options.
+   * @param okHttpClient            A {@link OkHttpClient} to use to communicate with the remote ILP-over-HTTP
+   *                                endpoint.
+   * @param objectMapper            A {@link ObjectMapper} for reading error responses from the remote ILP-over-HTTP
+   *                                endpoint.
+   * @param ilpCodecContext         A {@link CodecContext} for ILP.
+   * @param bearerTokenSupplier     A {@link BearerTokenSupplier} that can be used to get a bearer token to make
+   *                                authenticated calls to the remote HTTP endpoint.
+   */
+  public IlpOverHttpLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier,
+      final IlpOverHttpLinkSettings ilpOverHttpLinkSettings,
+      final OkHttpClient okHttpClient,
+      final ObjectMapper objectMapper,
+      final CodecContext ilpCodecContext,
+      final BearerTokenSupplier bearerTokenSupplier
+  ) {
+    super(operatorAddressSupplier, ilpOverHttpLinkSettings);
+    this.okHttpClient = Objects.requireNonNull(okHttpClient);
+    this.objectMapper = Objects.requireNonNull(objectMapper);
+    this.ilpCodecContext = Objects.requireNonNull(ilpCodecContext);
+    this.authTokenSupplier = Objects.requireNonNull(bearerTokenSupplier);
+  }
+
+  @Override
+  public InterledgerResponsePacket sendPacket(final InterledgerPreparePacket preparePacket) {
+    Objects.requireNonNull(preparePacket);
+
+    final Request okHttpRequest;
+    try {
+      final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+      ilpCodecContext.write(preparePacket, byteArrayOutputStream);
+      okHttpRequest = this.constructSendPacketRequest(preparePacket);
+    } catch (IOException e) {
+      throw new LinkException(e.getMessage(), e, getLinkId());
+    }
+
+    try (Response response = okHttpClient.newCall(okHttpRequest).execute()) {
+      if (response.isSuccessful()) {
+        //////////
+        // Success
+        return ilpCodecContext.read(InterledgerResponsePacket.class, response.body().byteStream());
+      } else {
+        //////////
+        // Reject!
+        final String errorResponseBody = CharStreams.toString(response.body().charStream());
+        Optional<ThrowableProblem> problem = parseThrowableProblem(preparePacket, errorResponseBody);
+        final InterledgerRejectPacket rejectPacket;
+
+        if (response.code() == 401 || response.code() == 403) {
+          // If this code is returned, we know the Link is misconfigured, so throw a LinkException Exception.
+          throw new LinkException(String.format(
+              "Unable to connect to remote ILP-over-HTTP Link: Invalid Bearer Token. response=%s", response
+          ), this.getLinkId());
+        } else if (response.code() >= 400 && response.code() < 500) {
+          // The request was bad for some reason, likely due to whatever is in the packet.
+          rejectPacket = InterledgerRejectPacket.builder()
+              .triggeredBy(getOperatorAddressSupplier().get())
+              .code(InterledgerErrorCode.F00_BAD_REQUEST)
+              .message(problem.map(Problem::getTitle).orElse(errorResponseBody))
+              .build();
+        } else {
+          // Something else went wrong on the server...try again later.
+          rejectPacket = InterledgerRejectPacket.builder()
+              .triggeredBy(getOperatorAddressSupplier().get())
+              .code(InterledgerErrorCode.T00_INTERNAL_ERROR)
+              .message(problem.map(Problem::getTitle).orElse(errorResponseBody))
+              .build();
+        }
+
+        logger.error(
+            "Unable to send ILP-over-HTTP packet. preparePacket={} httpResponseCode={} errorResponse={} "
+                + "rejectPacket={}",
+            preparePacket, response.code(), CharStreams.toString(response.body().charStream()), rejectPacket
+        );
+        return rejectPacket;
+      }
+    } catch (IOException e) {
+      throw new LinkException(e.getMessage(), e, getLinkId());
+    }
+  }
+
+  /**
+   * <p>Check the `/ilp` endpoint for ping by making an HTTP Head request with a ping packet, and
+   * asserting the values returned are one of the supported content-types required for ILP-over-HTTP.</p>
+   *
+   * <p>If the endpoint does not support producing ILP-over-HTTP responses, we expect a 406 NOT_ACCEPTABLE response. If
+   * the endpoint does not support ILP-over-HTTP requests, then we expect a 415 UNSUPPORTED_MEDIA_TYPE.</p>
+   */
+  public void testConnection() {
+    final OutgoingLinkSettings outgoingLinkSettings = this.getLinkSettings().outgoingHttpLinkSettings();
+    final String tokenSubject = outgoingLinkSettings.tokenSubject();
+    try {
+
+      final Request okHttpRequest = this.constructSendPacketRequest(UNFULFILLABLE_PACKET);
+
+      try (Response response = okHttpClient.newCall(okHttpRequest).execute()) {
+        if (response.isSuccessful()) {
+          //////////
+          // Success
+          // If there's one Accept header we can work with, then treat this as a successful test connection...
+          boolean responseHasOctetStreamContentType = response.headers(CONTENT_TYPE).stream()
+              .findFirst()
+              .filter(OCTET_STREAM_STRING::equals)
+              .isPresent();
+
+          if (responseHasOctetStreamContentType) {
+            logger.info("Remote peer-link supports ILP-over-HTTP. tokenSubject={} url={} responseHeaders={}",
+                outgoingLinkSettings.tokenSubject(), outgoingLinkSettings.url(), response
+            );
+          } else {
+            logger.warn("Remote peer-link supports ILP-over-HTTP but uses wrong ContentType. tokenSubject={} url={} "
+                    + "response={}",
+                outgoingLinkSettings.tokenSubject(), outgoingLinkSettings.url(), response
+            );
+          }
+        } else {
+          if (response.code() == 406 || response.code() == 415) { // NOT_ACCEPTABLE || UNSUPPORTED_MEDIA_TYPE
+            throw new LinkException(
+                String.format("Remote peer-link DOES NOT support ILP-over-HTTP. tokenSubject=%s url=%s response=%s",
+                    tokenSubject, outgoingLinkSettings.url(), response
+                ), getLinkId()
+            );
+          } else {
+            throw new LinkException(
+                String.format("Unable to connect to ILP-over-HTTP. tokenSubject=%s url=%s response=%s",
+                    tokenSubject, outgoingLinkSettings.url(), response
+                ), getLinkId()
+            );
+          }
+        }
+      }
+    } catch (IOException e) {
+      throw new LinkException(e.getMessage(), e, getLinkId());
+    }
+  }
+
+  /**
+   * Construct headers for an ILP-over-HTTP request.
+   *
+   * @return A newly constructed instance of {@link Headers}.
+   */
+  private Headers constructHttpRequestHeaders() {
+    final Headers.Builder headers = new Headers.Builder()
+        // Defaults to ILP_OCTET_STREAM, but is replaced by whatever testConnection returns if it's a valid media-type.
+        .add(HttpHeaders.ACCEPT, OCTET_STREAM.toString())
+        .add(CONTENT_TYPE, OCTET_STREAM.toString())
+        // Disable HTTP Caching of packets...
+        .add(CACHE_CONTROL, "private, max-age=0, no-cache")
+        .add(PRAGMA, "no-cache");
+
+    // Set the Operator Address header, if present.
+    getOperatorAddressSupplier().get().ifPresent(
+        operatorAddress -> headers
+            .set(IlpOverHttpConstants.ILP_OPERATOR_ADDRESS_VALUE, operatorAddress.getValue()));
+
+    headers.add(HttpHeaders.AUTHORIZATION, BEARER + this.authTokenSupplier.get());
+
+    return headers.build();
+  }
+
+  /**
+   * Helper method that attempts to parse a JSON error response into a {@link ThrowableProblem}.
+   *
+   * @param preparePacket     An {@link InterledgerPreparePacket}, for logging purposes.
+   * @param errorResponseBody An error {@link String} from the counter party ILP-over-HTTP endpoint.
+   *
+   * @return An optionally-present {@link ThrowableProblem} if one can be parsed from the {@link Response}.
+   */
+  private Optional<ThrowableProblem> parseThrowableProblem(
+      final InterledgerPreparePacket preparePacket, final String errorResponseBody
+  ) {
+    Objects.requireNonNull(preparePacket);
+    Objects.requireNonNull(errorResponseBody);
+
+    try {
+      return Optional.ofNullable(objectMapper.readValue(errorResponseBody, ThrowableProblem.class));
+    } catch (IOException e) {
+      // Something went wrong unmarshalling the error response. Return both the exception and the actual remote
+      // error.
+      logger.warn(
+          "Unable to parse Ilp-over-Http response. preparePacket={} errorResponseBody={} exception={}",
+          preparePacket, errorResponseBody, e
+      );
+      return Optional.empty();
+    }
+  }
+
+  /**
+   * Helper method to construct a {@link Request} containing the supplied {@code preparePacket}.
+   *
+   * @param preparePacket A {@link InterledgerPreparePacket} to send to the remote HTTP endpoint.
+   *
+   * @return A {@link Request} that can be used with an OkHttp client.
+   */
+  private Request constructSendPacketRequest(final InterledgerPreparePacket preparePacket) {
+    Objects.requireNonNull(preparePacket);
+
+    try {
+      final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+      ilpCodecContext.write(preparePacket, byteArrayOutputStream);
+
+      return new Builder()
+          .headers(constructHttpRequestHeaders())
+          .url(this.getLinkSettings().outgoingHttpLinkSettings().url())
+          .post(
+              RequestBody.create(byteArrayOutputStream.toByteArray(), APPLICATION_OCTET_STREAM)
+          )
+          .build();
+
+    } catch (Exception e) {
+      throw new LinkException(e.getMessage(), e, getLinkId());
+    }
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkFactory.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkFactory.java
new file mode 100644
index 0000000..000bbd9
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkFactory.java
@@ -0,0 +1,106 @@
+package org.interledger.link.http;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.encoding.asn.framework.CodecContext;
+import org.interledger.link.Link;
+import org.interledger.link.LinkFactory;
+import org.interledger.link.LinkSettings;
+import org.interledger.link.LinkType;
+import org.interledger.link.events.LinkConnectionEventEmitter;
+import org.interledger.link.http.auth.BearerTokenSupplier;
+import org.interledger.link.http.auth.Decryptor;
+import org.interledger.link.http.auth.JwtHs256BearerTokenSupplier;
+import org.interledger.link.http.auth.SharedSecretBytesSupplier;
+import org.interledger.link.http.auth.SimpleBearerTokenSupplier;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.OkHttpClient;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * An implementation of {@link LinkFactory} for creating Ilp-over-Http Links.
+ */
+public class IlpOverHttpLinkFactory implements LinkFactory {
+
+  private final LinkConnectionEventEmitter linkConnectionEventEmitter;
+  private final OkHttpClient okHttpClient;
+  private final Decryptor decryptor;
+  private final ObjectMapper objectMapper;
+  private final CodecContext ilpCodecContext;
+
+  public IlpOverHttpLinkFactory(
+      final LinkConnectionEventEmitter linkConnectionEventEmitter, final OkHttpClient okHttpClient,
+      final Decryptor decryptor,
+      final ObjectMapper objectMapper, final CodecContext ilpCodecContext
+  ) {
+    this.linkConnectionEventEmitter = Objects.requireNonNull(linkConnectionEventEmitter);
+    this.okHttpClient = Objects.requireNonNull(okHttpClient);
+    this.decryptor = Objects.requireNonNull(decryptor);
+    this.objectMapper = Objects.requireNonNull(objectMapper);
+    this.ilpCodecContext = Objects.requireNonNull(ilpCodecContext);
+  }
+
+  /**
+   * Construct a new instance of {@link Link} using the supplied inputs.
+   *
+   * @return A newly constructed instance of {@link Link}.
+   */
+  public Link<?> constructLink(
+      final Supplier<Optional<InterledgerAddress>> operatorAddressSupplier, final LinkSettings linkSettings
+  ) {
+    Objects.requireNonNull(linkSettings);
+
+    if (!this.supports(linkSettings.getLinkType())) {
+      throw new RuntimeException(
+          String.format("LinkType `%s` not supported by this factory!", linkSettings.getLinkType())
+      );
+    }
+
+    // Translate from Link.customSettings, being sure to apply custom settings from the incoming link.
+    final ImmutableIlpOverHttpLinkSettings.Builder builder = IlpOverHttpLinkSettings.builder().from(linkSettings);
+    final IlpOverHttpLinkSettings ilpOverHttpLinkSettings =
+        IlpOverHttpLinkSettings.applyCustomSettings(builder, linkSettings.getCustomSettings()).build();
+
+    final BearerTokenSupplier bearerTokenSupplier;
+    if (ilpOverHttpLinkSettings.outgoingHttpLinkSettings().authType().equals(IlpOverHttpLinkSettings.AuthType.SIMPLE)) {
+      // Decrypt whatever is inside of the encryptedTokenSharedSecret. For the SIMPLE profile, this will decrypt to the
+      // actual bearer token.
+      bearerTokenSupplier = new SimpleBearerTokenSupplier(new String(
+          decryptor.decrypt(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().encryptedTokenSharedSecret().getBytes())
+      ));
+    } else {
+      // TODO: For now, we assume the bytes are a String that conform to the Crypt CLI. However, this should be made
+      // type-safe and more generic if possible. E.g., CryptoCLI formate vs Protobuf. Or, standardize on a single
+      // type-safe format?
+
+      // NOTE: This supplier will always create a copy of the decrypted bytes so that the consumer of each call can
+      // safely wipe the bytes from memory without affecting other callers.
+      final SharedSecretBytesSupplier sharedSecretSupplier = () -> decryptor
+          .decrypt(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().encryptedTokenSharedSecret().getBytes());
+
+      bearerTokenSupplier = new JwtHs256BearerTokenSupplier(
+          sharedSecretSupplier, ilpOverHttpLinkSettings.outgoingHttpLinkSettings()
+      );
+    }
+
+    final IlpOverHttpLink ilpOverHttpLink = new IlpOverHttpLink(
+        operatorAddressSupplier,
+        ModifiableIlpOverHttpLinkSettings.create().from(ilpOverHttpLinkSettings), // Modifiable for testing
+        okHttpClient,
+        objectMapper,
+        ilpCodecContext,
+        bearerTokenSupplier
+    );
+
+    return ilpOverHttpLink;
+  }
+
+  @Override
+  public boolean supports(LinkType linkType) {
+    return IlpOverHttpLink.LINK_TYPE.equals(linkType);
+  }
+
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkSettings.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkSettings.java
new file mode 100644
index 0000000..8cd4136
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLinkSettings.java
@@ -0,0 +1,130 @@
+package org.interledger.link.http;
+
+import org.interledger.link.LinkSettings;
+import org.interledger.link.LinkType;
+
+import org.immutables.value.Value;
+import org.immutables.value.Value.Derived;
+import org.immutables.value.Value.Modifiable;
+
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * An extension of {@link LinkSettings} for ILP-over-HTTP links.
+ */
+public interface IlpOverHttpLinkSettings extends LinkSettings {
+
+  String DOT = ".";
+  String ILP_OVER_HTTP = "ilpOverHttp";
+
+  String OUTGOING = "outgoing";
+  String INCOMING = "incoming";
+
+  String AUTH_TYPE = "auth_type";
+
+  String TOKEN_ISSUER = "token_issuer";
+  String TOKEN_AUDIENCE = "token_audience";
+  String TOKEN_SUBJECT = "token_subject";
+  String TOKEN_EXPIRY = "token_expiry";
+
+  // Used to grab the auth credential from custom settings...
+  String SHARED_SECRET = "shared_secret";
+  String URL = "url";
+
+  static ImmutableIlpOverHttpLinkSettings.Builder builder() {
+    return ImmutableIlpOverHttpLinkSettings.builder();
+  }
+
+  /**
+   * Constructs a new builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param customSettings A {@link Map} of custom settings.
+   *
+   * @return A {@link ImmutableIlpOverHttpLinkSettings.Builder}.
+   */
+  static ImmutableIlpOverHttpLinkSettings.Builder fromCustomSettings(final Map<String, Object> customSettings) {
+    Objects.requireNonNull(customSettings);
+    return applyCustomSettings(IlpOverHttpLinkSettings.builder(), customSettings);
+  }
+
+  /**
+   * Populate a supplied builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param builder        A {@link ImmutableIlpOverHttpLinkSettings.Builder} to update with custom settings.
+   * @param customSettings A {@link Map} of custom settings.
+   *
+   * @return A {@link ImmutableIlpOverHttpLinkSettings.Builder}.
+   */
+  static ImmutableIlpOverHttpLinkSettings.Builder applyCustomSettings(
+      final ImmutableIlpOverHttpLinkSettings.Builder builder, Map<String, Object> customSettings
+  ) {
+    Objects.requireNonNull(builder);
+    Objects.requireNonNull(customSettings);
+
+    final ImmutableIncomingLinkSettings.Builder incomingLinkSettingsBuilder =
+        IncomingLinkSettings.fromCustomSettings(customSettings);
+    final ImmutableOutgoingLinkSettings.Builder outgoingLinkSettingsBuilder =
+        OutgoingLinkSettings.fromCustomSettings(customSettings);
+
+    builder.incomingHttpLinkSettings(incomingLinkSettingsBuilder.build());
+    builder.outgoingHttpLinkSettings(outgoingLinkSettingsBuilder.build());
+
+    builder.customSettings(customSettings);
+
+    return builder;
+  }
+
+  @Override
+  default LinkType getLinkType() {
+    return IlpOverHttpLink.LINK_TYPE;
+  }
+
+  /**
+   * Link settings for the incoming HTTP link.
+   *
+   * @return A {@link IncomingLinkSettings}.
+   */
+  IncomingLinkSettings incomingHttpLinkSettings();
+
+  /**
+   * Link settings for the outgoing HTTP link.
+   *
+   * @return A {@link OutgoingLinkSettings}.
+   */
+  OutgoingLinkSettings outgoingHttpLinkSettings();
+
+  /**
+   * <p>Defines currently-supported ILP-over-HTTP authentication profiles.</p>
+   *
+   * @see "https://github.com/interledger/rfcs/blob/master/0035-ilp-over-http/0035-ilp-over-http.md"
+   */
+  enum AuthType {
+    /**
+     * <p>The actual incoming and outgoing shared-secrets are used as Bearer tokens in an HTTP Authorization
+     * header.</p>
+     */
+    SIMPLE,
+
+    /**
+     * Use shared-secret symmetric keys to create and verify JWT_HS_256 tokens.
+     */
+    JWT_HS_256,
+
+    /**
+     * Use RSA asymmetric keys to create aand verify JWT_RS_256 tokens.
+     */
+    //JWT_RS_256
+  }
+
+  @Value.Immutable
+  @Modifiable
+  abstract class AbstractIlpOverHttpLinkSettings implements IlpOverHttpLinkSettings {
+
+    @Derived
+    public LinkType getLinkType() {
+      return IlpOverHttpLink.LINK_TYPE;
+    }
+
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IncomingLinkSettings.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IncomingLinkSettings.java
new file mode 100644
index 0000000..18db367
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IncomingLinkSettings.java
@@ -0,0 +1,146 @@
+package org.interledger.link.http;
+
+import static org.interledger.link.http.IlpOverHttpLinkSettings.AUTH_TYPE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.DOT;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.ILP_OVER_HTTP;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.INCOMING;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.SHARED_SECRET;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_AUDIENCE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_ISSUER;
+
+import okhttp3.HttpUrl;
+import org.immutables.value.Value;
+import org.immutables.value.Value.Modifiable;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * An extension of {@link SharedSecretTokenSettings} for incoming ILP-over-HTTP Links. Note that this interface
+ * purposefully does not define a `sub` claim because the `accountId` for any given account should always be used
+ * instead for verification purposes.
+ */
+public interface IncomingLinkSettings extends SharedSecretTokenSettings {
+
+  String HTTP_INCOMING_AUTH_TYPE = ILP_OVER_HTTP + DOT + INCOMING + DOT + AUTH_TYPE;
+
+  String HTTP_INCOMING_TOKEN_ISSUER = ILP_OVER_HTTP + DOT + INCOMING + DOT + TOKEN_ISSUER;
+  String HTTP_INCOMING_TOKEN_AUDIENCE = ILP_OVER_HTTP + DOT + INCOMING + DOT + TOKEN_AUDIENCE;
+  String HTTP_INCOMING_SHARED_SECRET = ILP_OVER_HTTP + DOT + INCOMING + DOT + SHARED_SECRET;
+
+  static ImmutableIncomingLinkSettings.Builder builder() {
+    return ImmutableIncomingLinkSettings.builder();
+  }
+
+  /**
+   * Constructs a new builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param customSettings A {@link Map} of custom settings to apply.
+   *
+   * @return A {@link ImmutableIncomingLinkSettings.Builder} with applied custom settings.
+   */
+  static ImmutableIncomingLinkSettings.Builder fromCustomSettings(Map<String, Object> customSettings) {
+    Objects.requireNonNull(customSettings);
+
+    return applyCustomSettings(ImmutableIncomingLinkSettings.builder(), customSettings);
+  }
+
+  /**
+   * Populate a supplied builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param builder        A {@link ImmutableIncomingLinkSettings.Builder} to apply custom settings to.
+   * @param customSettings A {@link Map} of custom settings to apply.
+   *
+   * @return A {@link ImmutableIncomingLinkSettings.Builder} with applied custom settings.
+   */
+  static ImmutableIncomingLinkSettings.Builder applyCustomSettings(
+      final ImmutableIncomingLinkSettings.Builder builder, Map<String, Object> customSettings
+  ) {
+    Objects.requireNonNull(builder);
+    Objects.requireNonNull(customSettings);
+
+    // When loaded from a properties file, the properties are hierarchical in a Map. However, in Java, they are not, so
+    // consider both options. Generally only one will be present, but if for some reason both are present, the String
+    // values will win.
+    Optional.ofNullable(customSettings.get(ILP_OVER_HTTP))
+        .map(val -> (Map<String, Object>) val)
+        .ifPresent(blastSettings -> Optional.ofNullable(blastSettings.get(INCOMING))
+            .map(val -> (Map<String, Object>) val)
+            .ifPresent(incomingSettings -> {
+
+              Optional.ofNullable(incomingSettings.get(AUTH_TYPE))
+                  .map(Object::toString)
+                  .map(String::toUpperCase)
+                  .map(IlpOverHttpLinkSettings.AuthType::valueOf)
+                  .ifPresent(builder::authType);
+
+              Optional.ofNullable(incomingSettings.get(SHARED_SECRET))
+                  .map(Object::toString)
+                  .ifPresent(builder::encryptedTokenSharedSecret);
+
+              Optional.ofNullable(incomingSettings.get(TOKEN_ISSUER))
+                  .map(Object::toString)
+                  .map(HttpUrl::parse)
+                  .ifPresent(builder::tokenIssuer);
+
+              Optional.ofNullable(incomingSettings.get(TOKEN_AUDIENCE))
+                  .map(Object::toString)
+                  .map(HttpUrl::parse)
+                  .ifPresent(builder::tokenAudience);
+            }));
+
+    Optional.ofNullable(customSettings.get(HTTP_INCOMING_AUTH_TYPE))
+        .map(Object::toString)
+        .map(String::toUpperCase)
+        .map(IlpOverHttpLinkSettings.AuthType::valueOf)
+        .ifPresent(builder::authType);
+
+    Optional.ofNullable(customSettings.get(HTTP_INCOMING_TOKEN_ISSUER))
+        .map(Object::toString)
+        .map(HttpUrl::parse)
+        .ifPresent(builder::tokenIssuer);
+
+    Optional.ofNullable(customSettings.get(HTTP_INCOMING_TOKEN_AUDIENCE))
+        .map(Object::toString)
+        .map(HttpUrl::parse)
+        .ifPresent(builder::tokenAudience);
+
+    Optional.ofNullable(customSettings.get(HTTP_INCOMING_SHARED_SECRET))
+        .map(Object::toString)
+        .ifPresent(builder::encryptedTokenSharedSecret);
+
+    return builder;
+  }
+
+  /**
+   * <p>The minimum amount of time (in milliseconds) to budget for receiving a response message from an account.</p>
+   *
+   * <p>Especially useful for ILP packets, if a packet expires in 30 seconds, then a link should only wait 29 seconds
+   * before timing out so that it can generally be sure to reject the request (as opposed to merely allowing a timeout
+   * to occur, because timeouts are ambiguous).</p>
+   *
+   * @return A {@link Duration}.
+   */
+  default Duration getMinMessageWindow() {
+    return Duration.of(2500, ChronoUnit.MILLIS);
+  }
+
+  @Value.Immutable
+  @Modifiable
+  abstract class AbstractIncomingLinkSettings implements IncomingLinkSettings {
+
+    @Override
+    @Value.Default
+    public Duration getMinMessageWindow() {
+      return Duration.of(2500, ChronoUnit.MILLIS);
+    }
+
+    @Override
+    @Value.Redacted
+    public abstract String encryptedTokenSharedSecret();
+
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/OutgoingLinkSettings.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/OutgoingLinkSettings.java
new file mode 100644
index 0000000..9ce30ea
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/OutgoingLinkSettings.java
@@ -0,0 +1,183 @@
+package org.interledger.link.http;
+
+import static org.interledger.link.http.IlpOverHttpLinkSettings.AUTH_TYPE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.DOT;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.ILP_OVER_HTTP;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.OUTGOING;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.SHARED_SECRET;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_AUDIENCE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_EXPIRY;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_ISSUER;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_SUBJECT;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.URL;
+
+import okhttp3.HttpUrl;
+import org.immutables.value.Value;
+import org.immutables.value.Value.Modifiable;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+public interface OutgoingLinkSettings extends SharedSecretTokenSettings {
+
+  String HTTP_OUTGOING_AUTH_TYPE = ILP_OVER_HTTP + DOT + OUTGOING + DOT + AUTH_TYPE;
+
+  String HTTP_OUTGOING_TOKEN_ISSUER = ILP_OVER_HTTP + DOT + OUTGOING + DOT + TOKEN_ISSUER;
+  String HTTP_OUTGOING_TOKEN_AUDIENCE = ILP_OVER_HTTP + DOT + OUTGOING + DOT + TOKEN_AUDIENCE;
+  String HTTP_OUTGOING_TOKEN_SUBJECT = ILP_OVER_HTTP + DOT + OUTGOING + DOT + TOKEN_SUBJECT;
+  String HTTP_OUTGOING_TOKEN_EXPIRY = ILP_OVER_HTTP + DOT + OUTGOING + DOT + TOKEN_EXPIRY;
+
+  String HTTP_OUTGOING_SHARED_SECRET = ILP_OVER_HTTP + DOT + OUTGOING + DOT + SHARED_SECRET;
+  String HTTP_OUTGOING_URL = ILP_OVER_HTTP + DOT + OUTGOING + DOT + URL;
+
+  static ImmutableOutgoingLinkSettings.Builder builder() {
+    return ImmutableOutgoingLinkSettings.builder();
+  }
+
+  /**
+   * Constructs a new builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param customSettings A {@link Map} of custom settings to apply.
+   *
+   * @return A {@link ImmutableOutgoingLinkSettings.Builder} with applied custom settings.
+   */
+  static ImmutableOutgoingLinkSettings.Builder fromCustomSettings(Map<String, Object> customSettings) {
+    Objects.requireNonNull(customSettings);
+
+    return applyCustomSettings(ImmutableOutgoingLinkSettings.builder(), customSettings);
+  }
+
+  /**
+   * Populate a supplied builder with the correct custom settings, as found in {@code customSettings}.
+   *
+   * @param builder        A {@link ImmutableIncomingLinkSettings.Builder} to apply custom settings to.
+   * @param customSettings A {@link Map} of custom settings to apply.
+   *
+   * @return A {@link ImmutableOutgoingLinkSettings.Builder} with applied custom settings.
+   */
+  static ImmutableOutgoingLinkSettings.Builder applyCustomSettings(
+      final ImmutableOutgoingLinkSettings.Builder builder, Map<String, Object> customSettings
+  ) {
+    Objects.requireNonNull(builder);
+    Objects.requireNonNull(customSettings);
+
+    // When loaded from a properties file, the properties are hierarchical in a Map. However, in Java, they are not, so
+    // consider both options. Generally only one will be present, but if for some reason both are present, the String
+    // values will win.
+    Optional.ofNullable(customSettings.get(ILP_OVER_HTTP))
+        .map(val -> (Map<String, Object>) val)
+        .ifPresent(blastSettings -> {
+
+          Optional.ofNullable(blastSettings.get(OUTGOING))
+              .map(val -> (Map<String, Object>) val)
+              .ifPresent(outgoingSettings -> {
+
+                Optional.ofNullable(outgoingSettings.get(TOKEN_SUBJECT))
+                    .map(Object::toString)
+                    .ifPresent(builder::tokenSubject);
+
+                Optional.ofNullable(outgoingSettings.get(TOKEN_ISSUER))
+                    .map(Object::toString)
+                    .map(HttpUrl::parse)
+                    .ifPresent(builder::tokenIssuer);
+
+                Optional.ofNullable(outgoingSettings.get(TOKEN_AUDIENCE))
+                    .map(Object::toString)
+                    .map(HttpUrl::parse)
+                    .ifPresent(builder::tokenAudience);
+
+                Optional.ofNullable(outgoingSettings.get(AUTH_TYPE))
+                    .map(Object::toString)
+                    .map(String::toUpperCase)
+                    .map(IlpOverHttpLinkSettings.AuthType::valueOf)
+                    .ifPresent(builder::authType);
+
+                Optional.ofNullable(outgoingSettings.get(SHARED_SECRET))
+                    .map(Object::toString)
+                    .ifPresent(builder::encryptedTokenSharedSecret);
+
+                Optional.ofNullable(outgoingSettings.get(TOKEN_EXPIRY))
+                    .map(Object::toString)
+                    .map(Duration::parse)
+                    .ifPresent(builder::tokenExpiry);
+
+                Optional.ofNullable(outgoingSettings.get(URL))
+                    .map(Object::toString)
+                    .map(HttpUrl::parse)
+                    .ifPresent(builder::url);
+
+              });
+        });
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_TOKEN_ISSUER))
+        .map(Object::toString)
+        .map(HttpUrl::parse)
+        .ifPresent(builder::tokenIssuer);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_TOKEN_AUDIENCE))
+        .map(Object::toString)
+        .map(HttpUrl::parse)
+        .ifPresent(builder::tokenAudience);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_AUTH_TYPE))
+        .map(Object::toString)
+        .map(String::toUpperCase)
+        .map(IlpOverHttpLinkSettings.AuthType::valueOf)
+        .ifPresent(builder::authType);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_TOKEN_SUBJECT))
+        .map(Object::toString)
+        .ifPresent(builder::tokenSubject);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_SHARED_SECRET))
+        .map(Object::toString)
+        .ifPresent(builder::encryptedTokenSharedSecret);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_TOKEN_EXPIRY))
+        .map(Object::toString)
+        .map(Duration::parse)
+        .ifPresent(builder::tokenExpiry);
+
+    Optional.ofNullable(customSettings.get(HTTP_OUTGOING_URL))
+        .map(Object::toString)
+        .map(HttpUrl::parse)
+        .ifPresent(builder::url);
+
+    return builder;
+  }
+
+  /**
+   * The expected `sub` value of an ILP-over-HTTP authentication token. This values identifies the principal that the
+   * token is authenticating.
+   *
+   * @return A {@link String} with the token subject.
+   */
+  String tokenSubject();
+
+  /**
+   * If present, determines how often to sign a new token for auth. Optional to support the shared-secret use-case.
+   *
+   * @return An optionally present {@link Duration} representing the token expiry.
+   */
+  Optional<Duration> tokenExpiry();
+
+  /**
+   * endpoint to POST packets to. If url contains a percent and the link is in `multi` mode, then the segment after this
+   * link's own address will be filled where the `%` is  when routing packets.
+   *
+   * @return An {@link HttpUrl} for this Link.
+   */
+  HttpUrl url();
+
+  @Value.Immutable
+  @Modifiable
+  abstract class AbstractOutgoingLinkSettings implements OutgoingLinkSettings {
+
+    @Override
+    @Value.Redacted
+    public abstract String encryptedTokenSharedSecret();
+
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/SharedSecretTokenSettings.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/SharedSecretTokenSettings.java
new file mode 100644
index 0000000..a00ee24
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/SharedSecretTokenSettings.java
@@ -0,0 +1,45 @@
+package org.interledger.link.http;
+
+import okhttp3.HttpUrl;
+
+import java.util.Optional;
+
+/**
+ * Defines settings for a link that uses JWT_HS_256 with HMAC-SHA256 bearer tokens employing a shared secret.
+ */
+public interface SharedSecretTokenSettings {
+
+  /**
+   * The type of Auth to support for outgoing HTTP connections.
+   *
+   * @return A {@link IlpOverHttpLinkSettings.AuthType} for this shared secret token settings.
+   */
+  IlpOverHttpLinkSettings.AuthType authType();
+
+  /**
+   * The expected `iss` value of the issuer of a Blast token. This value should always be a URL so that it can be rooted
+   * in Internet PKI and compared against the TLS certificate issued by the other side of a blast connection (i.e., the
+   * remote peer). It is optional in order to support node-wide issuance, or account-level issuance.
+   *
+   * @return An optionally present {@link HttpUrl} for the token issuer.
+   */
+  Optional<HttpUrl> tokenIssuer();
+
+  /**
+   * The expected `aud` claim value of an incoming JWT_HS_256 token. In general, this value should be the URL of the
+   * Connector operating this BLAST link, since the remote will want to narrow the scope of its token to only be valid
+   * on this endpoint.
+   *
+   * @return An optionally present {@link HttpUrl} for the token audience.
+   */
+  Optional<HttpUrl> tokenAudience();
+
+  /**
+   * An encrypted shared `secret`, encoded per `EncryptedSecret`, that can be used to authenticate an incoming
+   * ILP-over-HTTP (BLAST) connection.
+   *
+   * @return A {@link String} containing the shared secret in base64 encoding.
+   */
+  String encryptedTokenSharedSecret();
+
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/BearerTokenSupplier.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/BearerTokenSupplier.java
new file mode 100644
index 0000000..b7a3e28
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/BearerTokenSupplier.java
@@ -0,0 +1,24 @@
+package org.interledger.link.http.auth;
+
+import java.util.function.Supplier;
+
+/**
+ * <p>Defines a {@link Supplier} that provides a JWT that can be used as an ILP-over-HTTP bearer token.</p>
+ *
+ * <p>This implementation attempts to reduce the amount of time the shared-secret is in-memory by preemptively
+ * generating an authentication token that conforms to `JWT_HS_256`, and then wiping the shared-secret from it allocated
+ * memory. Additionally, generated auth tokens have short durations (i.e, they expire), so typing them as Strings is
+ * tolerable for this implementation.</p>
+ */
+public interface BearerTokenSupplier extends Supplier<String> {
+
+  /**
+   * Gets an ILP-over-HTTP bearer token which can be used to make requests to a remote HTTP endpoint supporting
+   * ILP-over-HTTP.
+   *
+   * @return A JWT bearer token, as a {@link String}.
+   */
+  @Override
+  String get();
+
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/Decryptor.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/Decryptor.java
new file mode 100644
index 0000000..9d2f33e
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/Decryptor.java
@@ -0,0 +1,21 @@
+package org.interledger.link.http.auth;
+
+/**
+ * <p>Defines a generic interface for decrypting ciphertext values.</p>
+ *
+ * <p>A primary use-case for this interface is to decrypt properties or values that are sitting in-memory in encrypted
+ * form. For example, an encrypted shared-secret signing key may have been loaded from a property file or a datastore.
+ * This interface allows implementations to define how to decrypt such information using arbitrary schemes.</p>
+ */
+public interface Decryptor {
+
+  /**
+   * Load an actual secret using the supplied information.
+   *
+   * @param cipherMessage The byte[] array containing cipherText and other meta-data than can be used to decrypt the
+   *                      ciphertext.
+   *
+   * @return A byte array containing the unencrypted secret value, in binary form.
+   */
+  byte[] decrypt(final byte[] cipherMessage);
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/JwtHs256BearerTokenSupplier.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/JwtHs256BearerTokenSupplier.java
new file mode 100644
index 0000000..5264255
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/JwtHs256BearerTokenSupplier.java
@@ -0,0 +1,105 @@
+package org.interledger.link.http.auth;
+
+import org.interledger.link.http.OutgoingLinkSettings;
+
+import com.auth0.jwt.JWT;
+import com.auth0.jwt.algorithms.Algorithm;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+
+/**
+ * <p>Defines a {@link Supplier} that provides a JWT that can be used as an ILP-over-HTTP bearer token.</p>
+ *
+ * <p>This implementation attempts to reduce the amount of time the shared-secret is in-memory by preemptively
+ * generating an authentication token that conforms to `JWT_HS_256`, and then wiping the shared-secret from it allocated
+ * memory. Additionally, generated auth tokens have short durations (i.e, they expire), so typing them as Strings is
+ * tolerable for this implementation.</p>
+ */
+public class JwtHs256BearerTokenSupplier implements BearerTokenSupplier {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  private final LoadingCache<String, String> ilpOverHttpAuthTokens;
+  private final OutgoingLinkSettings outgoingLinkSettings;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param sharedSecretBytesSupplier A {@link SharedSecretBytesSupplier} that returns a copy of the shared secret used
+   *                                  to sign JWTs using the HS_256 algorithm.
+   * @param outgoingLinkSettings      A {@link OutgoingLinkSettings} that contains all settings required to construct a
+   *                                  bearer token that can auth an outgoing request.
+   */
+  public JwtHs256BearerTokenSupplier(
+      final SharedSecretBytesSupplier sharedSecretBytesSupplier,
+      final OutgoingLinkSettings outgoingLinkSettings
+  ) {
+    Objects.requireNonNull(sharedSecretBytesSupplier);
+    Objects.requireNonNull(outgoingLinkSettings);
+
+    this.outgoingLinkSettings = Objects.requireNonNull(outgoingLinkSettings);
+
+    ilpOverHttpAuthTokens = CacheBuilder.newBuilder()
+        // There should only ever be 1 or 2 tokens in-memory for a given client instance.
+        .maximumSize(3)
+        // Expire after this duration, which will correspond to the last incoming request from the peer.
+        .expireAfterAccess(this.outgoingLinkSettings.tokenExpiry().orElse(Duration.of(30, ChronoUnit.MINUTES)))
+        .removalListener((RemovalListener<String, String>) notification ->
+            logger.debug("Removing IlpOverHttp AuthToken from Cache for Principal: {}", notification.getKey())
+        )
+        .build(new CacheLoader<String, String>() {
+          public String load(final String authSubject) {
+            Objects.requireNonNull(authSubject);
+
+            final byte[] sharedSecretBytes = sharedSecretBytesSupplier.get();
+            try {
+              return JWT.create()
+                  //.withIssuedAt(new Date())
+                  //      .withIssuer(getOutgoingLinkSettings()
+                  //        .tokenIssuer()
+                  //        .map(HttpUrl::toString)
+                  //        .orElseThrow(() -> new RuntimeException("JWT Blast Senders require an Outgoing Issuer!"))
+                  //      )
+                  .withSubject(outgoingLinkSettings.tokenSubject()) // account identifier at the remote server.
+                  // Expire at the appointed time, or else after 15 minutes.
+                  .withExpiresAt(JwtHs256BearerTokenSupplier.this.outgoingLinkSettings.tokenExpiry()
+                      .map(expiry -> Date.from(Instant.now().plus(expiry)))
+                      .orElseGet(() -> Date.from(Instant.now().plus(15, ChronoUnit.MINUTES))))
+                  .sign(Algorithm.HMAC256(sharedSecretBytes));
+            } finally {
+              // Zero-out all bytes in the `sharedSecretBytes` array.
+              Arrays.fill(sharedSecretBytes, (byte) 0);
+            }
+          }
+        });
+  }
+
+  /**
+   * Gets an ILP-over-HTTP bearer token which can be used to make requests to a remote HTTP endpoint supporting
+   * ILP-over-HTTP.
+   *
+   * @return A JWT bearer token, as a {@link String}.
+   */
+  @Override
+  public String get() {
+    try {
+      return this.ilpOverHttpAuthTokens.get(outgoingLinkSettings.tokenSubject());
+    } catch (ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SharedSecretBytesSupplier.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SharedSecretBytesSupplier.java
new file mode 100644
index 0000000..a5a5c29
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SharedSecretBytesSupplier.java
@@ -0,0 +1,26 @@
+package org.interledger.link.http.auth;
+
+import java.util.function.Supplier;
+
+/**
+ * <p>Defines a {@link Supplier} that provides a byte-array containing the shared-secret that can be used to perform
+ * JWT signing operations. Note that this interface is generally targeted at the "HMAC with SHA-2" family of algorithms
+ * defined in RFC-7518.</p>
+ *
+ * <p>This implementation attempts to reduce the amount of time the shared-secret is in-memory by preemptively
+ * generating an authentication token that conforms to `JWT_HS_256`, copying it, and then wiping the shared-secret from
+ * it allocated memory. Consumers of the secret bytes should take care to do the same.</p>
+ *
+ * @see "https://tools.ietf.org/html/rfc7518#section-3.2"
+ */
+public interface SharedSecretBytesSupplier extends Supplier<byte[]> {
+
+  /**
+   * Return a shared secret that can be used to sign JWT tokens using the HS_256 alogrithm.
+   *
+   * @return A JWT bearer token, as a {@link String}.
+   */
+  @Override
+  byte[] get();
+
+}
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SimpleBearerTokenSupplier.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SimpleBearerTokenSupplier.java
new file mode 100644
index 0000000..38b084e
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/auth/SimpleBearerTokenSupplier.java
@@ -0,0 +1,42 @@
+package org.interledger.link.http.auth;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * <p>Defines a {@link Supplier} that provides a plain-text bearer auth token.</p>
+ *
+ * <p>WARNING: This implementation is intended for testing and debugging purposes, and should not be used in a
+ * production environment because the tokens produced never expire, and are held in memory. Instead, consider using
+ * {@link JwtHs256BearerTokenSupplier} instead.</p>
+ */
+public class SimpleBearerTokenSupplier implements BearerTokenSupplier {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  private final String bearerAuthToken;
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param bearerAuthToken An array of chars that contains a plaintext bearer auth token.
+   */
+  public SimpleBearerTokenSupplier(final String bearerAuthToken) {
+    this.bearerAuthToken = Objects.requireNonNull(bearerAuthToken);
+  }
+
+  /**
+   * Gets an ILP-over-HTTP bearer token which can be used to make requests to a remote HTTP endpoint supporting
+   * ILP-over-HTTP.
+   *
+   * @return A simple plain-text bearer token, as a {@link String}.
+   */
+  @Override
+  public String get() {
+    return bearerAuthToken;
+  }
+
+}
diff --git a/link-parent/pom.xml b/link-parent/pom.xml
new file mode 100644
index 0000000..0d691aa
--- /dev/null
+++ b/link-parent/pom.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>quilt-parent</artifactId>
+    <groupId>org.interledger</groupId>
+    <version>0.18.0-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <name>Quilt :: Link Parent</name>
+  <artifactId>link-parent</artifactId>
+  <packaging>pom</packaging>
+
+  <modules>
+    <module>link-core</module>
+    <module>link-ilp-over-http</module>
+  </modules>
+
+</project>
diff --git a/pom.xml b/pom.xml
index 4d1b899..97a7c5f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -98,6 +98,7 @@
     <module>ilp-core</module>
     <module>btp-core</module>
     <module>ildcp-core</module>
+    <module>link-parent</module>
     <module>stream-parent</module>
     <module>jackson-datatypes</module>
     <module>codecs-parent</module>
@@ -136,6 +137,18 @@
         <version>${project.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>${project.groupId}</groupId>
+        <artifactId>link-core</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>${project.groupId}</groupId>
+        <artifactId>link-ilp-over-http</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>${project.groupId}</groupId>
         <artifactId>stream-core</artifactId>
@@ -177,6 +190,23 @@
         <artifactId>ildcp-core-codecs</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.auth0</groupId>
+        <artifactId>java-jwt</artifactId>
+        <version>3.8.2</version>
+      </dependency>
+
+      <dependency>
+        <groupId>com.auth0</groupId>
+        <artifactId>jwks-rsa</artifactId>
+        <version>0.8.3</version>
+      </dependency>
+
+      <dependency>
+        <groupId>com.squareup.okhttp3</groupId>
+        <artifactId>okhttp</artifactId>
+        <version>4.1.0</version>
+      </dependency>
 
       <dependency>
         <groupId>com.fasterxml.jackson.core</groupId>
@@ -255,6 +285,24 @@
         <scope>test</scope>
       </dependency>
 
+      <dependency>
+        <groupId>org.slf4j</groupId>
+        <artifactId>slf4j-api</artifactId>
+        <version>1.7.28</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.zalando</groupId>
+        <artifactId>jackson-datatype-problem</artifactId>
+        <version>0.23.0</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.zalando</groupId>
+        <artifactId>problem</artifactId>
+        <version>0.23.0</version>
+      </dependency>
+
     </dependencies>
   </dependencyManagement>
 


Test patch:
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/AbstractLinkTest.java b/link-parent/link-core/src/test/java/org/interledger/link/AbstractLinkTest.java
new file mode 100644
index 0000000..c5e583f
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/AbstractLinkTest.java
@@ -0,0 +1,134 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * Unit tests for {@link AbstractLink}.
+ */
+public class AbstractLinkTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.operator");
+  private static final LinkId LINK_ID = LinkId.of("foo");
+  private static final LinkSettings LINK_SETTINGS = LinkSettings.builder()
+      .linkType(LoopbackLink.LINK_TYPE)
+      .build();
+
+  private AbstractLink<?> link;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+
+    this.link = new TestAbstractLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LINK_SETTINGS
+    );
+    this.link.setLinkId(LINK_ID);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void getLinkIdWhenNull() {
+    this.link = new TestAbstractLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LinkSettings.builder().linkType(LinkType.of("foo")).build()
+    );
+    try {
+      link.getLinkId();
+    } catch (IllegalStateException e) {
+      assertThat(e.getMessage()).isEqualTo("The LinkId must be set before using a Link");
+      throw e;
+    }
+  }
+
+  @Test
+  public void getLinkId() {
+    assertThat(link.getLinkId()).isEqualTo(LINK_ID);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void setLinkIdWhenAlreadySet() {
+    try {
+      link.setLinkId(LinkId.of("bar"));
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e.getMessage()).isEqualTo("LinkId may only be set once");
+      throw e;
+    }
+  }
+
+  @Test
+  public void getOperatorAddressSupplier() {
+    assertThat(link.getOperatorAddressSupplier().get().get()).isEqualTo(OPERATOR_ADDRESS);
+  }
+
+  @Test
+  public void getLinkSettings() {
+    assertThat(link.getLinkSettings()).isEqualTo(LINK_SETTINGS);
+  }
+
+  @Test
+  public void registerAndUnregisterLinkHandler() {
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+
+    LinkHandler linkHandlerMock = mock(LinkHandler.class);
+    link.registerLinkHandler(linkHandlerMock);
+    assertThat(link.getLinkHandler().get()).isEqualTo(linkHandlerMock);
+
+    link.unregisterLinkHandler();
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void safeGetLinkHandlerWithoutRegistering() {
+    try {
+      link.safeGetLinkHandler();
+      fail();
+    } catch (Exception e) {
+      assertThat(e.getMessage()).isEqualTo("You MUST register a LinkHandler before using this link!");
+      throw e;
+    }
+  }
+
+  @Test
+  public void safeGetLinkHandler() {
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+    LinkHandler linkHandlerMock = mock(LinkHandler.class);
+    link.registerLinkHandler(linkHandlerMock);
+    assertThat(link.safeGetLinkHandler()).isEqualTo(linkHandlerMock);
+  }
+
+  /**
+   * Helper class for testing {@link AbstractLink}.
+   */
+  private class TestAbstractLink extends AbstractLink {
+
+    private TestAbstractLink(Supplier operatorAddressSupplier, LinkSettings linkSettings) {
+      super(operatorAddressSupplier, linkSettings);
+    }
+
+    @Override
+    public InterledgerResponsePacket sendPacket(InterledgerPreparePacket preparePacket) {
+      return InterledgerRejectPacket.builder()
+          .triggeredBy(OPERATOR_ADDRESS)
+          .code(InterledgerErrorCode.F08_AMOUNT_TOO_LARGE)
+          .message("the message")
+          .build();
+    }
+  }
+
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/AbstractStatefulLinkTest.java b/link-parent/link-core/src/test/java/org/interledger/link/AbstractStatefulLinkTest.java
new file mode 100644
index 0000000..248b94d
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/AbstractStatefulLinkTest.java
@@ -0,0 +1,360 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
+import org.interledger.link.AbstractStatefulLink.EventBusConnectionEventEmitter;
+import org.interledger.link.events.LinkConnectedEvent;
+import org.interledger.link.events.LinkConnectionEventEmitter;
+import org.interledger.link.events.LinkConnectionEventListener;
+import org.interledger.link.events.LinkDisconnectedEvent;
+
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+
+/**
+ * Unit tests for {@link AbstractStatefulLink}.
+ */
+public class AbstractStatefulLinkTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.operator");
+  private static final LinkId LINK_ID = LinkId.of("foo");
+  private static final LinkSettings LINK_SETTINGS = LinkSettings.builder()
+      .linkType(LoopbackLink.LINK_TYPE)
+      .build();
+
+  @Mock
+  private LinkConnectionEventEmitter linkConnectionEventEmitterMock;
+
+  private AbstractStatefulLink<?> link;
+
+  private AtomicBoolean doConnectCalled;
+  private AtomicBoolean doDisconnectCalled;
+
+  private AtomicBoolean onConnectCalled;
+  private AtomicBoolean onDisconnectCalled;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+
+    this.doConnectCalled = new AtomicBoolean();
+    this.doDisconnectCalled = new AtomicBoolean();
+    this.onConnectCalled = new AtomicBoolean();
+    this.onDisconnectCalled = new AtomicBoolean();
+
+    link = new TestAbstractStatefulLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LINK_SETTINGS,
+        linkConnectionEventEmitterMock
+    );
+    link.setLinkId(LINK_ID);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void getLinkIdWhenNull() {
+    link = new TestAbstractStatefulLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LINK_SETTINGS,
+        linkConnectionEventEmitterMock
+    );
+
+    try {
+      link.getLinkId();
+    } catch (IllegalStateException e) {
+      assertThat(e.getMessage()).isEqualTo("The LinkId must be set before using a Link");
+      throw e;
+    }
+  }
+
+  @Test
+  public void getLinkId() {
+    assertThat(link.getLinkId()).isEqualTo(LINK_ID);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void setLinkIdWhenAlreadySet() {
+    try {
+      link.setLinkId(LinkId.of("bar"));
+      fail();
+    } catch (IllegalStateException e) {
+      assertThat(e.getMessage()).isEqualTo("LinkId may only be set once");
+      throw e;
+    }
+  }
+
+  @Test
+  public void getOperatorAddressSupplier() {
+    assertThat(link.getOperatorAddressSupplier().get().get()).isEqualTo(OPERATOR_ADDRESS);
+  }
+
+  @Test
+  public void getLinkSettings() {
+    assertThat(link.getLinkSettings()).isEqualTo(LINK_SETTINGS);
+  }
+
+  @Test
+  public void connect() {
+    assertThat(doConnectCalled.get()).isEqualTo(false);
+    assertThat(link.isConnected()).isEqualTo(false);
+    link.connect();
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(link.isConnected()).isEqualTo(true);
+    verify(linkConnectionEventEmitterMock).emitEvent(Mockito.<LinkConnectedEvent>any());
+    verifyNoMoreInteractions(linkConnectionEventEmitterMock);
+  }
+
+  @Test
+  public void connectAndDisconnect() {
+    assertThat(doConnectCalled.get()).isEqualTo(false);
+    assertThat(doDisconnectCalled.get()).isEqualTo(false);
+    assertThat(link.isConnected()).isEqualTo(false);
+
+    link.connect();
+
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(doDisconnectCalled.get()).isEqualTo(false);
+    assertThat(link.isConnected()).isEqualTo(true);
+
+    link.disconnect();
+
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(doDisconnectCalled.get()).isEqualTo(true);
+    assertThat(link.isConnected()).isEqualTo(false);
+  }
+
+  @Test
+  public void close() {
+    assertThat(doConnectCalled.get()).isEqualTo(false);
+    assertThat(link.isConnected()).isEqualTo(false);
+    link.connect();
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(link.isConnected()).isEqualTo(true);
+
+    link.close();
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(doDisconnectCalled.get()).isEqualTo(true);
+    assertThat(link.isConnected()).isEqualTo(false);
+  }
+
+  @Test
+  public void disconnect() {
+    assertThat(doDisconnectCalled.get()).isEqualTo(false);
+    assertThat(link.isConnected()).isEqualTo(false);
+    link.connect();
+    assertThat(doConnectCalled.get()).isEqualTo(true);
+    assertThat(link.isConnected()).isEqualTo(true);
+  }
+
+  @Test
+  public void registerAndUnregisterLinkHandler() {
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+
+    LinkHandler linkHandlerMock = mock(LinkHandler.class);
+    link.registerLinkHandler(linkHandlerMock);
+    assertThat(link.getLinkHandler().get()).isEqualTo(linkHandlerMock);
+
+    link.unregisterLinkHandler();
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void safeGetLinkHandlerWithoutRegistering() {
+    try {
+      link.safeGetLinkHandler();
+      fail();
+    } catch (Exception e) {
+      assertThat(e.getMessage()).isEqualTo("You MUST register a LinkHandler before using this link!");
+      throw e;
+    }
+  }
+
+  @Test
+  public void safeGetLinkHandler() {
+    assertThat(link.getLinkHandler().isPresent()).isEqualTo(false);
+    LinkHandler linkHandlerMock = mock(LinkHandler.class);
+    link.registerLinkHandler(linkHandlerMock);
+    assertThat(link.safeGetLinkHandler()).isEqualTo(linkHandlerMock);
+  }
+
+  @Test
+  public void addLinkEventListener() {
+    final LinkConnectionEventListener linkConnectionEventListenerMock = new LinkConnectionEventListener() {
+
+      @Override
+      @Subscribe
+      public void onConnect(LinkConnectedEvent event) {
+        onConnectCalled.set(true);
+      }
+
+      @Override
+      @Subscribe
+      public void onDisconnect(LinkDisconnectedEvent event) {
+        onDisconnectCalled.set(true);
+      }
+    };
+
+    final EventBus eventBus = new EventBus();
+
+    this.link = new TestAbstractStatefulLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LINK_SETTINGS,
+        new EventBusConnectionEventEmitter(eventBus)
+    );
+
+    link.setLinkId(LINK_ID);
+    link.addLinkEventListener(linkConnectionEventListenerMock);
+
+    assertThat(onConnectCalled.get()).isEqualTo(false);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+
+    link.connect();
+    assertThat(onConnectCalled.get()).isEqualTo(true);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+
+    link.sendPacket(mock(InterledgerPreparePacket.class));
+    assertThat(onConnectCalled.get()).isEqualTo(true);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+
+    link.disconnect();
+    assertThat(onConnectCalled.get()).isEqualTo(true);
+    assertThat(onDisconnectCalled.get()).isEqualTo(true);
+
+    // Try sending when disconnected.
+    link.sendPacket(mock(InterledgerPreparePacket.class));
+    assertThat(onConnectCalled.get()).isEqualTo(true);
+    assertThat(onDisconnectCalled.get()).isEqualTo(true);
+  }
+
+  @Test
+  public void sendWhenDisconnected() {
+    LinkConnectionEventListener linkConnectionEventListenerMock = mock(LinkConnectionEventListener.class);
+    link.addLinkEventListener(linkConnectionEventListenerMock);
+
+    // Try sending when disconnected.
+    link.sendPacket(mock(InterledgerPreparePacket.class));
+    assertThat(onConnectCalled.get()).isEqualTo(false);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+  }
+
+  @Test
+  public void onConnectionError() {
+    final LinkConnectionEventListener linkConnectionEventListener = new LinkConnectionEventListener() {
+      @Override
+      public void onConnect(LinkConnectedEvent event) {
+      }
+
+      @Override
+      public void onDisconnect(LinkDisconnectedEvent event) {
+      }
+    };
+
+    final EventBus eventBus = new EventBus();
+
+    link = new AbstractStatefulLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LINK_SETTINGS,
+        new EventBusConnectionEventEmitter(eventBus)
+    ) {
+      @Override
+      public CompletableFuture<Void> doConnect() {
+        doConnectCalled.set(true);
+        return CompletableFuture.supplyAsync(() -> {
+          throw new RuntimeException("connect error");
+        });
+      }
+
+      @Override
+      public CompletableFuture<Void> doDisconnect() {
+        doDisconnectCalled.set(true);
+        return CompletableFuture.completedFuture(null);
+      }
+
+      @Override
+      public InterledgerResponsePacket sendPacket(InterledgerPreparePacket preparePacket) {
+        return mock(InterledgerResponsePacket.class);
+      }
+    };
+
+    link.addLinkEventListener(linkConnectionEventListener);
+
+    assertThat(onConnectCalled.get()).isEqualTo(false);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+
+    link.setLinkId(LINK_ID);
+
+    try {
+      link.connect().join();
+      fail();
+    } catch (Exception e) {
+      assertThat(e.getMessage()).isEqualTo("java.lang.RuntimeException: connect error");
+    }
+
+    assertThat(onConnectCalled.get()).isEqualTo(false);
+    assertThat(onDisconnectCalled.get()).isEqualTo(false);
+  }
+
+  @Test
+  public void removeLinkEventListener() {
+    LinkConnectionEventListener linkConnectionEventListenerMock = mock(LinkConnectionEventListener.class);
+    link.addLinkEventListener(linkConnectionEventListenerMock);
+    link.removeLinkEventListener(linkConnectionEventListenerMock);
+
+    link.connect();
+    link.disconnect();
+    verifyNoMoreInteractions(linkConnectionEventListenerMock);
+  }
+
+  /**
+   * Helper class for testing {@link AbstractStatefulLink}.
+   */
+  private class TestAbstractStatefulLink extends AbstractStatefulLink {
+
+    private TestAbstractStatefulLink(
+        Supplier operatorAddressSupplier, LinkSettings linkSettings,
+        LinkConnectionEventEmitter linkConnectionEventEmitter
+    ) {
+      super(operatorAddressSupplier, linkSettings, linkConnectionEventEmitter);
+    }
+
+    @Override
+    public CompletableFuture<Void> doConnect() {
+      doConnectCalled.set(true);
+      return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> doDisconnect() {
+      doDisconnectCalled.set(true);
+      return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public InterledgerResponsePacket sendPacket(InterledgerPreparePacket preparePacket) {
+      return InterledgerRejectPacket.builder()
+          .triggeredBy(OPERATOR_ADDRESS)
+          .code(InterledgerErrorCode.F08_AMOUNT_TOO_LARGE)
+          .message("the message")
+          .build();
+    }
+  }
+
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/LinkFactoryProviderTest.java b/link-parent/link-core/src/test/java/org/interledger/link/LinkFactoryProviderTest.java
new file mode 100644
index 0000000..e4f0131
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/LinkFactoryProviderTest.java
@@ -0,0 +1,65 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.interledger.link.exceptions.LinkException;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Unit test for {@link LinkFactoryProvider}.
+ */
+public class LinkFactoryProviderTest {
+
+  public static final LinkType TEST_LINK_TYPE = LinkType.of("foo");
+
+  @Mock
+  private LinkFactory linkFactoryMock;
+
+  private LinkFactoryProvider linkFactoryProvider;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+    this.linkFactoryProvider = new LinkFactoryProvider();
+  }
+
+  @Test
+  public void registerAndGetLinkFactory() {
+    linkFactoryProvider.registerLinkFactory(TEST_LINK_TYPE, linkFactoryMock);
+    assertThat(linkFactoryProvider.getLinkFactory(TEST_LINK_TYPE)).isEqualTo(linkFactoryMock);
+  }
+
+  @Test(expected = LinkException.class)
+  public void getUnregisteredLinkFactory() {
+    try {
+      linkFactoryProvider.getLinkFactory(TEST_LINK_TYPE);
+    } catch (LinkException e) {
+      assertThat(e.getMessage()).isEqualTo("No registered LinkFactory linkType=LinkType(FOO)");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void registerLinkFactoryWithNullLinkType() {
+    try {
+      linkFactoryProvider.registerLinkFactory(null, linkFactoryMock);
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("linkType must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void registerLinkFactoryWithNullLinkFactory() {
+    try {
+      linkFactoryProvider.registerLinkFactory(TEST_LINK_TYPE, null);
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("linkFactory must not be null");
+      throw e;
+    }
+  }
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/LinkSettingsTest.java b/link-parent/link-core/src/test/java/org/interledger/link/LinkSettingsTest.java
new file mode 100644
index 0000000..dde2be3
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/LinkSettingsTest.java
@@ -0,0 +1,23 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link LinkSettings}.
+ */
+public class LinkSettingsTest {
+
+  @Test
+  public void builder() {
+    final LinkSettings linkSettings = LinkSettings.builder()
+        .linkType(LinkType.of("foo"))
+        .putCustomSettings("foo", "bar")
+        .build();
+
+    assertThat(linkSettings.getLinkType()).isEqualTo(LinkType.of("foo"));
+    assertThat(linkSettings.getCustomSettings().size()).isEqualTo(1);
+    assertThat(linkSettings.getCustomSettings().get("foo")).isEqualTo("bar");
+  }
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkFactoryTest.java b/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkFactoryTest.java
new file mode 100644
index 0000000..952c064
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkFactoryTest.java
@@ -0,0 +1,101 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.exceptions.LinkException;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Optional;
+
+/**
+ * Unit tests for {@link LoopbackLinkFactory}.
+ */
+public class LoopbackLinkFactoryTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.operator");
+  private final LinkId linkId = LinkId.of("foo");
+
+  @Mock
+  private LinkSettings linkSettingsMock;
+
+  @Mock
+  private PacketRejector packetRejectorMock;
+
+
+  private LoopbackLinkFactory loopbackLinkFactory;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+    this.loopbackLinkFactory = new LoopbackLinkFactory(packetRejectorMock);
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void constructWithNulPacketRejector() {
+    try {
+      new LoopbackLinkFactory(null);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("packetRejector must not be null");
+      throw e;
+    }
+  }
+
+  @Test
+  public void supports() {
+    assertThat(loopbackLinkFactory.supports(LoopbackLink.LINK_TYPE)).isEqualTo(true);
+    assertThat(loopbackLinkFactory.supports(LinkType.of("foo"))).isEqualTo(false);
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void constructLinkWithNullOperator() {
+    try {
+      loopbackLinkFactory.constructLink(null, linkSettingsMock);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("operatorAddressSupplier must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void constructLinkWithNullLinkSettings() {
+    try {
+      loopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), null);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("linkSettings must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = LinkException.class)
+  public void constructLinkWithUnsupportedLinkType() {
+    try {
+      LinkSettings linkSettings = LinkSettings.builder()
+          .linkType(LinkType.of("foo"))
+          .build();
+      loopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), linkSettings);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("LinkType not supported by this factory. linkType=LinkType(FOO)");
+      throw e;
+    }
+  }
+
+  @Test
+  public void constructLink() {
+    LinkSettings linkSettings = LinkSettings.builder()
+        .linkType(LoopbackLink.LINK_TYPE)
+        .build();
+    Link<?> link = loopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), linkSettings);
+    link.setLinkId(linkId);
+    assertThat(link.getLinkId()).isEqualTo(linkId);
+  }
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkTest.java b/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkTest.java
new file mode 100644
index 0000000..1e12838
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/LoopbackLinkTest.java
@@ -0,0 +1,149 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.interledger.link.LoopbackLink.SIMULATED_REJECT_ERROR_CODE;
+import static org.junit.Assert.fail;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerConstants;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+
+import com.google.common.collect.Maps;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Unit tests for {@link LoopbackLink}.
+ */
+public class LoopbackLinkTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.foo");
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  private PacketRejector packetRejector;
+
+  private LoopbackLink link;
+
+  @Before
+  public void setUp() {
+    this.packetRejector = new PacketRejector(() -> Optional.of(OPERATOR_ADDRESS));
+
+    this.link = new LoopbackLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LinkSettings.builder().linkType(LoopbackLink.LINK_TYPE).build(),
+        packetRejector
+    );
+  }
+
+  @Test(expected = RuntimeException.class)
+  public void registerLinkHandler() {
+    try {
+      link.registerLinkHandler(incomingPreparePacket -> null);
+    } catch (Exception e) {
+      assertThat(e.getMessage())
+          .isEqualTo("Loopback links never have incoming data, and thus should not have a registered DataHandler.");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void sendPacketWithNull() {
+    try {
+      link.sendPacket(null);
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("preparePacket must not be null");
+      throw e;
+    }
+  }
+
+  @Test
+  public void sendPacket() {
+    this.link = new LoopbackLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LinkSettings.builder().linkType(LoopbackLink.LINK_TYPE).build(),
+        packetRejector
+    );
+    link.setLinkId(LinkId.of("foo"));
+
+    final InterledgerPreparePacket preparePacket = preparePacket();
+    link.sendPacket(preparePacket()).handle(
+        fulfillPacket -> {
+          assertThat(fulfillPacket.getFulfillment()).isEqualTo(LoopbackLink.LOOPBACK_FULFILLMENT);
+          assertThat(fulfillPacket.getData()).isEqualTo(preparePacket.getData());
+        },
+        rejectPacket -> {
+          logger.error("rejectPacket={}", rejectPacket);
+          fail("Expected a Fulfill");
+        }
+    );
+  }
+
+  @Test
+  public void sendT02Packet() {
+    final Map<String, String> customSettings = Maps.newHashMap();
+    customSettings.put(SIMULATED_REJECT_ERROR_CODE, InterledgerErrorCode.T02_PEER_BUSY_CODE);
+    this.link = new LoopbackLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LinkSettings.builder().linkType(LoopbackLink.LINK_TYPE).customSettings(customSettings).build(),
+        packetRejector
+    );
+    link.setLinkId(LinkId.of("foo"));
+
+    link.sendPacket(preparePacket()).handle(
+        fulfillPacket -> {
+          logger.error("interledgerRejectPacket={}", fulfillPacket);
+          fail("Expected a Reject");
+        },
+        rejectPacket -> {
+          assertThat(rejectPacket.getCode()).isEqualTo(InterledgerErrorCode.T02_PEER_BUSY);
+        }
+    );
+  }
+
+  @Test(expected = RuntimeException.class)
+  public void sendT99Packet() {
+    final Map<String, String> customSettings = Maps.newHashMap();
+    customSettings.put(SIMULATED_REJECT_ERROR_CODE, InterledgerErrorCode.T99_APPLICATION_ERROR_CODE);
+    this.link = new LoopbackLink(
+        () -> Optional.of(OPERATOR_ADDRESS),
+        LinkSettings.builder().linkType(LoopbackLink.LINK_TYPE).customSettings(customSettings).build(),
+        packetRejector
+    );
+    link.setLinkId(LinkId.of("foo"));
+
+    try {
+      link.sendPacket(preparePacket()).handle(
+          fulfillPacket -> {
+            logger.error("interledgerRejectPacket={}", fulfillPacket);
+            fail("Expected a Reject");
+          },
+          rejectPacket -> {
+            logger.error("interledgerRejectPacket={}", rejectPacket);
+            fail("Expected a Reject");
+          }
+      );
+    } catch (RuntimeException e) {
+      assertThat(e.getMessage()).isEqualTo("T99 APPLICATION ERROR");
+      throw e;
+    }
+  }
+
+  private InterledgerPreparePacket preparePacket() {
+    return InterledgerPreparePacket.builder()
+        .amount(BigInteger.TEN)
+        .executionCondition(InterledgerConstants.ALL_ZEROS_CONDITION)
+        .destination(OPERATOR_ADDRESS)
+        .expiresAt(Instant.now())
+        .data(new byte[32])
+        .build();
+  }
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/PacketRejectorTest.java b/link-parent/link-core/src/test/java/org/interledger/link/PacketRejectorTest.java
new file mode 100644
index 0000000..99e9273
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/PacketRejectorTest.java
@@ -0,0 +1,105 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.math.BigInteger;
+import java.time.Instant;
+import java.util.Optional;
+
+/**
+ * Unit tests for {@link PacketRejector}.
+ */
+public class PacketRejectorTest {
+
+  private static final LinkId LINK_ID = LinkId.of("fooLink");
+  private static final InterledgerPreparePacket PREPARE_PACKET = InterledgerPreparePacket.builder()
+      .expiresAt(Instant.now())
+      .amount(BigInteger.ONE)
+      .destination(InterledgerAddress.of("test.destination"))
+      .executionCondition(PingLoopbackLink.PING_PROTOCOL_CONDITION)
+      .build();
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.operator");
+
+  private PacketRejector packetRejector;
+
+  @Before
+  public void setup() {
+    this.packetRejector = new PacketRejector(() -> Optional.of(OPERATOR_ADDRESS));
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void nullConstructor() {
+    try {
+      new PacketRejector(null);
+      fail();
+    } catch (Exception e) {
+      assertThat(e.getMessage()).isEqualTo("operatorAddressSupplier must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void nullLinkId() {
+    try {
+      packetRejector.reject(null, PREPARE_PACKET, InterledgerErrorCode.T02_PEER_BUSY, "");
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("rejectingLinkId must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void nullPreparePacket() {
+    try {
+      packetRejector.reject(LINK_ID, null, InterledgerErrorCode.T02_PEER_BUSY, "");
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("preparePacket must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void nullErrorCode() {
+    try {
+      packetRejector.reject(LINK_ID, PREPARE_PACKET, null, "");
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("errorCode must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void nullErrorMessage() {
+    try {
+      packetRejector.reject(LINK_ID, PREPARE_PACKET, InterledgerErrorCode.T02_PEER_BUSY, null);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("errorMessage must not be null");
+      throw e;
+    }
+  }
+
+  @Test
+  public void testReject() {
+    InterledgerRejectPacket reject = packetRejector
+        .reject(LINK_ID, PREPARE_PACKET, InterledgerErrorCode.T02_PEER_BUSY, "the error");
+
+    assertThat(reject.getCode()).isEqualTo(InterledgerErrorCode.T02_PEER_BUSY);
+    assertThat(reject.getTriggeredBy().get()).isEqualTo(OPERATOR_ADDRESS);
+    assertThat(reject.getMessage()).isEqualTo("the error");
+    assertThat(reject.getData().length).isEqualTo(0);
+  }
+
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkFactoryTest.java b/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkFactoryTest.java
new file mode 100644
index 0000000..fb896df
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkFactoryTest.java
@@ -0,0 +1,86 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.link.exceptions.LinkException;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Optional;
+
+/**
+ * Unit tests for {@link PingLoopbackLinkFactory}.
+ */
+public class PingLoopbackLinkFactoryTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.operator");
+  private final LinkId LINK_ID = LinkId.of("foo");
+
+  @Mock
+  private LinkSettings linkSettingsMock;
+
+  private PingLoopbackLinkFactory pingLoopbackLinkFactory;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+    this.pingLoopbackLinkFactory = new PingLoopbackLinkFactory();
+  }
+
+  @Test
+  public void supports() {
+    assertThat(pingLoopbackLinkFactory.supports(PingLoopbackLink.LINK_TYPE)).isEqualTo(true);
+    assertThat(pingLoopbackLinkFactory.supports(LinkType.of("foo"))).isEqualTo(false);
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void constructLinkWithNullOperator() {
+    try {
+      pingLoopbackLinkFactory.constructLink(null, linkSettingsMock);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("operatorAddressSupplier must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void constructLinkWithNullLinkSettings() {
+    try {
+      pingLoopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), null);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("linkSettings must not be null");
+      throw e;
+    }
+  }
+
+  @Test(expected = LinkException.class)
+  public void constructLinkWithUnsupportedLinkType() {
+    try {
+      LinkSettings linkSettings = LinkSettings.builder()
+          .linkType(LinkType.of("foo"))
+          .build();
+      pingLoopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), linkSettings);
+      fail();
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("LinkType not supported by this factory. linkType=LinkType(FOO)");
+      throw e;
+    }
+  }
+
+  @Test
+  public void constructLink() {
+    LinkSettings linkSettings = LinkSettings.builder()
+        .linkType(PingLoopbackLink.LINK_TYPE)
+        .build();
+    Link<?> link = pingLoopbackLinkFactory.constructLink(() -> Optional.of(OPERATOR_ADDRESS), linkSettings);
+    link.setLinkId(LINK_ID);
+    assertThat(link.getLinkId()).isEqualTo(LINK_ID);
+  }
+}
diff --git a/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkTest.java b/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkTest.java
new file mode 100644
index 0000000..897f008
--- /dev/null
+++ b/link-parent/link-core/src/test/java/org/interledger/link/PingLoopbackLinkTest.java
@@ -0,0 +1,167 @@
+package org.interledger.link;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.interledger.link.PingLoopbackLink.PING_PROTOCOL_CONDITION;
+import static org.interledger.link.PingLoopbackLink.PING_PROTOCOL_FULFILLMENT;
+import static org.junit.Assert.fail;
+
+import org.interledger.core.InterledgerAddress;
+import org.interledger.core.InterledgerCondition;
+import org.interledger.core.InterledgerErrorCode;
+import org.interledger.core.InterledgerFulfillment;
+import org.interledger.core.InterledgerPreparePacket;
+import org.interledger.core.InterledgerRejectPacket;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.Optional;
+
+/**
+ * Unit tests for {@link PingLoopbackLink}.
+ */
+public class PingLoopbackLinkTest {
+
+  private static final InterledgerAddress OPERATOR_ADDRESS = InterledgerAddress.of("test.foo");
+
+  private static final InterledgerRejectPacket EXPECTED_REJECT_PACKET = InterledgerRejectPacket.builder()
+      .triggeredBy(OPERATOR_ADDRESS)
+      .code(InterledgerErrorCode.F00_BAD_REQUEST)
+      .message("error message")
+      .build();
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+  @Mock
+  LinkSettings linkSettingsMock;
+
+  private PingLoopbackLink link;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+    this.link = new PingLoopbackLink(() -> Optional.of(OPERATOR_ADDRESS), linkSettingsMock);
+  }
+
+  @Test(expected = RuntimeException.class)
+  public void registerLinkHandler() {
+    try {
+      link.registerLinkHandler(incomingPreparePacket -> null);
+    } catch (Exception e) {
+      assertThat(e.getMessage()).isEqualTo(
+          "PingLoopback links never have incoming data, and thus should not have a registered DataHandler.");
+      throw e;
+    }
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void sendPacketWithNull() {
+    try {
+      link.sendPacket(null);
+    } catch (NullPointerException e) {
+      assertThat(e.getMessage()).isEqualTo("preparePacket must not be null!");
+      throw e;
+    }
+  }
+
+  @Test
+  public void testFulfillmentFromBase64() {
+    byte[] bytes = Base64.getDecoder().decode("cGluZ3BpbmdwaW5ncGluZ3BpbmdwaW5ncGluZ3Bpbmc=");
+    final InterledgerFulfillment expectedFulfillment = InterledgerFulfillment.of(bytes);
+
+    assertThat(expectedFulfillment).isEqualTo(PING_PROTOCOL_FULFILLMENT);
+    assertThat(expectedFulfillment.getCondition()).isEqualTo(PING_PROTOCOL_CONDITION);
+    assertThat(expectedFulfillment.validateCondition(PING_PROTOCOL_CONDITION)).isEqualTo(true);
+  }
+
+  @Test
+  public void testFulfillmentFromAscii() {
+    final InterledgerFulfillment expectedFulfillment =
+        InterledgerFulfillment.of("pingpingpingpingpingpingpingping".getBytes(StandardCharsets.US_ASCII));
+
+    assertThat(expectedFulfillment).isEqualTo(PING_PROTOCOL_FULFILLMENT);
+    assertThat(expectedFulfillment.getCondition()).isEqualTo(PING_PROTOCOL_CONDITION);
+    assertThat(expectedFulfillment.validateCondition(PING_PROTOCOL_CONDITION)).isEqualTo(true);
+  }
+
+  @Test
+  public void testConditionFromBase64() {
+    byte[] bytes = Base64.getDecoder().decode("jAC8DGFPZPfh4AtZpXuvXFe2oRmpDVSvSJg2oT+bx34=");
+    final InterledgerCondition expectedCondition = InterledgerCondition.of(bytes);
+
+    assertThat(expectedCondition).isEqualTo(PING_PROTOCOL_CONDITION);
+    assertThat(PING_PROTOCOL_FULFILLMENT.getCondition()).isEqualTo(PING_PROTOCOL_CONDITION);
+    assertThat(PING_PROTOCOL_FULFILLMENT.validateCondition(PING_PROTOCOL_CONDITION)).isEqualTo(true);
+  }
+
+  @Test
+  public void sendPacketWithInvalidCondition() {
+    final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
+        .amount(BigInteger.TEN)
+        .executionCondition(InterledgerCondition.of(new byte[32]))
+        .destination(OPERATOR_ADDRESS)
+        .expiresAt(Instant.now())
+        .build();
+
+    link.sendPacket(preparePacket).handle(fulfillPacket -> {
+          logger.error("InterledgerFulfillPacket: {}", fulfillPacket);
+          fail("Expected a Reject!");
+        },
+        rejectPacket -> {
+          assertThat(rejectPacket.getCode()).isEqualTo(InterledgerErrorCode.F00_BAD_REQUEST);
+          assertThat(rejectPacket.getTriggeredBy().isPresent()).isEqualTo(true);
+          assertThat(rejectPacket.getTriggeredBy()).isEqualTo(EXPECTED_REJECT_PACKET.getTriggeredBy());
+        }
+    );
+  }
+
+  /**
+   * A packet with an invalid destination address should never make its way into this link, so this tests expects a
+   * Fulfill even though that's counter-intuitive -- this is because the Link itself does not destination address
+   * checking.
+   */
+  @Test
+  public void sendPacketWithInvalidAddress() {
+    final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
+        .amount(BigInteger.TEN)
+        .executionCondition(PING_PROTOCOL_CONDITION)
+        .destination(InterledgerAddress.of("example.foo"))
+        .expiresAt(Instant.now())
+        .build();
+
+    link.sendPacket(preparePacket).handle(
+        fulfillPacket -> assertThat(fulfillPacket.getFulfillment()).isEqualTo(PING_PROTOCOL_FULFILLMENT),
+        rejectPacket -> {
+          logger.error("InterledgerRejectPacket: {}", rejectPacket);
+          fail("Expected a Fulfill!");
+        }
+    );
+  }
+
+  @Test
+  public void sendPacket() {
+    final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder()
+        .amount(BigInteger.TEN)
+        .executionCondition(PING_PROTOCOL_CONDITION)
+        .destination(OPERATOR_ADDRESS)
+        .expiresAt(Instant.now())
+        .build();
+
+    link.sendPacket(preparePacket).handle(
+        fulfillPacket -> assertThat(fulfillPacket.getFulfillment()).isEqualTo(PING_PROTOCOL_FULFILLMENT),
+        rejectPacket -> {
+          logger.error("InterledgerRejectPacket: {}", rejectPacket);
+          fail("Expected a Fulfill!");
+        }
+    );
+  }
+
+}
diff --git a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/AbstractHttpLinkSettingsTest.java b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/AbstractHttpLinkSettingsTest.java
new file mode 100644
index 0000000..31fea4d
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/AbstractHttpLinkSettingsTest.java
@@ -0,0 +1,78 @@
+package org.interledger.link.http;
+
+import static org.interledger.link.http.IlpOverHttpLinkSettings.AUTH_TYPE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.ILP_OVER_HTTP;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.INCOMING;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.OUTGOING;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.SHARED_SECRET;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_AUDIENCE;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_EXPIRY;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_ISSUER;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.TOKEN_SUBJECT;
+import static org.interledger.link.http.IlpOverHttpLinkSettings.URL;
+import static org.interledger.link.http.IncomingLinkSettings.HTTP_INCOMING_AUTH_TYPE;
+import static org.interledger.link.http.IncomingLinkSettings.HTTP_INCOMING_SHARED_SECRET;
+import static org.interledger.link.http.IncomingLinkSettings.HTTP_INCOMING_TOKEN_AUDIENCE;
+import static org.interledger.link.http.IncomingLinkSettings.HTTP_INCOMING_TOKEN_ISSUER;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_AUTH_TYPE;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_SHARED_SECRET;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_TOKEN_AUDIENCE;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_TOKEN_EXPIRY;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_TOKEN_ISSUER;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_TOKEN_SUBJECT;
+import static org.interledger.link.http.OutgoingLinkSettings.HTTP_OUTGOING_URL;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+public abstract class AbstractHttpLinkSettingsTest {
+
+  protected Map<String, Object> customSettingsFlat() {
+    return ImmutableMap.<String, Object>builder()
+        .put(HTTP_INCOMING_AUTH_TYPE, IlpOverHttpLinkSettings.AuthType.JWT_HS_256.name())
+        .put(HTTP_INCOMING_TOKEN_ISSUER, "https://incoming-issuer.example.com/")
+        .put(HTTP_INCOMING_SHARED_SECRET, "incoming-credential")
+        .put(HTTP_INCOMING_TOKEN_AUDIENCE, "https://incoming-audience.example.com/")
+
+        .put(HTTP_OUTGOING_AUTH_TYPE, IlpOverHttpLinkSettings.AuthType.SIMPLE.name())
+        .put(HTTP_OUTGOING_TOKEN_SUBJECT, "outgoing-subject")
+        .put(HTTP_OUTGOING_SHARED_SECRET, "outgoing-credential")
+        .put(HTTP_OUTGOING_TOKEN_ISSUER, "https://outgoing-issuer.example.com/")
+        .put(HTTP_OUTGOING_TOKEN_AUDIENCE, "https://outgoing-audience.example.com/")
+        .put(HTTP_OUTGOING_TOKEN_EXPIRY, Duration.ofDays(1).toString())
+        .put(HTTP_OUTGOING_URL, "https://outgoing.example.com")
+
+        .build();
+  }
+
+  protected Map<String, Object> customSettingsHeirarchical() {
+    final Map<String, Object> incomingMap = new HashMap<>();
+    incomingMap.put(AUTH_TYPE, IlpOverHttpLinkSettings.AuthType.JWT_HS_256.name());
+    incomingMap.put(TOKEN_SUBJECT, "incoming-subject");
+    incomingMap.put(SHARED_SECRET, "incoming-credential");
+    incomingMap.put(TOKEN_ISSUER, "https://incoming-issuer.example.com/");
+    incomingMap.put(TOKEN_AUDIENCE, "https://incoming-audience.example.com/");
+
+    final Map<String, Object> outgoingMap = new HashMap<>();
+    outgoingMap.put(AUTH_TYPE, IlpOverHttpLinkSettings.AuthType.SIMPLE.name());
+    outgoingMap.put(TOKEN_SUBJECT, "outgoing-subject");
+    outgoingMap.put(SHARED_SECRET, "outgoing-credential");
+    outgoingMap.put(TOKEN_ISSUER, "https://outgoing-issuer.example.com/");
+    outgoingMap.put(TOKEN_AUDIENCE, "https://outgoing-audience.example.com/");
+    outgoingMap.put(TOKEN_EXPIRY, Duration.ofDays(2).toString());
+    outgoingMap.put(URL, "https://outgoing.example.com/");
+
+    final Map<String, Object> httpMap = new HashMap<>();
+    httpMap.put(INCOMING, incomingMap);
+    httpMap.put(OUTGOING, outgoingMap);
+
+    final Map<String, Object> customSettings = new HashMap<>();
+    customSettings.put(ILP_OVER_HTTP, httpMap);
+
+    return customSettings;
+  }
+
+}
diff --git a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverIlpOverHttpLinkSettingsTest.java b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverIlpOverHttpLinkSettingsTest.java
new file mode 100644
index 0000000..4fd7204
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverIlpOverHttpLinkSettingsTest.java
@@ -0,0 +1,140 @@
+package org.interledger.link.http;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import okhttp3.HttpUrl;
+import org.junit.Test;
+
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Unit tests for {@link IlpOverHttpLinkSettings}.
+ */
+public class IlpOverIlpOverHttpLinkSettingsTest extends AbstractHttpLinkSettingsTest {
+
+  /**
+   * Tests the builder when customAttributes is a flat collection of key/value pairs using dotted-notation.
+   */
+  @Test
+  public void applyCustomSettingsWithFlatDottedNotation() {
+    final Map<String, Object> flattenedCustomSettings = this.customSettingsFlat();
+
+    final ImmutableIlpOverHttpLinkSettings.Builder builder = IlpOverHttpLinkSettings.builder();
+    final IlpOverHttpLinkSettings ilpOverHttpLinkSettings =
+        IlpOverHttpLinkSettings.applyCustomSettings(builder, flattenedCustomSettings).build();
+
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.JWT_HS_256);
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("incoming-credential");
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().getMinMessageWindow())
+        .isEqualTo(Duration.ofMillis(2500));
+
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("outgoing-credential");
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenExpiry().get()).isEqualTo(Duration.ofHours(24));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().url())
+        .isEqualTo(HttpUrl.parse("https://outgoing.example.com/"));
+  }
+
+  /**
+   * Tests the builder when customAttributes is a Map of Maps.
+   */
+  @Test
+  public void applyCustomSettingsWithMapHeirarchy() {
+    final Map<String, Object> customSettings = this.customSettingsHeirarchical();
+
+    final ImmutableIlpOverHttpLinkSettings.Builder builder = IlpOverHttpLinkSettings.builder();
+    final ImmutableIlpOverHttpLinkSettings httpLinkSettings =
+        IlpOverHttpLinkSettings.applyCustomSettings(builder, customSettings).build();
+
+    assertThat(httpLinkSettings.incomingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.JWT_HS_256);
+    assertThat(httpLinkSettings.incomingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(httpLinkSettings.incomingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(httpLinkSettings.incomingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("incoming-credential");
+    assertThat(httpLinkSettings.incomingHttpLinkSettings().getMinMessageWindow()).isEqualTo(Duration.ofMillis(2500));
+
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com/"));
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com/"));
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("outgoing-credential");
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().tokenExpiry().get()).isEqualTo(Duration.ofHours(48));
+    assertThat(httpLinkSettings.outgoingHttpLinkSettings().url())
+        .isEqualTo(HttpUrl.parse("https://outgoing.example.com/"));
+  }
+
+  @Test
+  public void testWithoutCustomSettings() {
+    final IncomingLinkSettings incomingLinksettings =
+        IncomingLinkSettings.builder()
+            .authType(IlpOverHttpLinkSettings.AuthType.SIMPLE)
+            .tokenIssuer(HttpUrl.parse("https://incoming-issuer.example.com/"))
+            .tokenAudience(HttpUrl.parse("https://incoming-audience.example.com/"))
+            .minMessageWindow(Duration.ofMillis(30))
+            .encryptedTokenSharedSecret("incoming-credential")
+            .build();
+
+    final OutgoingLinkSettings outgoingLinksettings =
+        OutgoingLinkSettings.builder()
+            .authType(IlpOverHttpLinkSettings.AuthType.SIMPLE)
+            .tokenSubject("outgoing-subject")
+            .tokenIssuer(HttpUrl.parse("https://outgoing-issuer.example.com/"))
+            .tokenAudience(HttpUrl.parse("https://outgoing-audience.example.com/"))
+            .encryptedTokenSharedSecret("outgoing-credential")
+            .tokenExpiry(Duration.ofMillis(40))
+            .url(HttpUrl.parse("https://outgoing.example.com/"))
+            .build();
+
+    final IlpOverHttpLinkSettings ilpOverHttpLinkSettings = IlpOverHttpLinkSettings.builder()
+        .incomingHttpLinkSettings(incomingLinksettings)
+        .outgoingHttpLinkSettings(outgoingLinksettings)
+        .build();
+
+    assertThat(ilpOverHttpLinkSettings.getLinkType()).isEqualTo(IlpOverHttpLink.LINK_TYPE);
+
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("incoming-credential");
+    assertThat(ilpOverHttpLinkSettings.incomingHttpLinkSettings().getMinMessageWindow())
+        .isEqualTo(Duration.ofMillis(30));
+
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().authType())
+        .isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com/"));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().encryptedTokenSharedSecret())
+        .isEqualTo("outgoing-credential");
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().tokenExpiry().get()).isEqualTo(Duration.ofMillis(40));
+    assertThat(ilpOverHttpLinkSettings.outgoingHttpLinkSettings().url())
+        .isEqualTo(HttpUrl.parse("https://outgoing.example.com/"));
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IncomingIlpOverIlpOverHttpLinkSettingsSettingsTest.java b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IncomingIlpOverIlpOverHttpLinkSettingsSettingsTest.java
new file mode 100644
index 0000000..0e6f626
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IncomingIlpOverIlpOverHttpLinkSettingsSettingsTest.java
@@ -0,0 +1,74 @@
+package org.interledger.link.http;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import okhttp3.HttpUrl;
+import org.junit.Test;
+
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Unit tests for {@link IncomingLinkSettings}.
+ */
+public class IncomingIlpOverIlpOverHttpLinkSettingsSettingsTest extends AbstractHttpLinkSettingsTest {
+
+  // This value doesn't _strictly_ need to be encrypted for purposes of this test. It could easily be plain-text, but
+  // for completeness we use the encrypted test-variant.
+  private static final String SHH
+      = "enc:JKS:crypto.p12:secret0:1:aes_gcm:AAAADKZPmASojt1iayb2bPy4D-Toq7TGLTN95HzCQAeJtz0=";
+
+  /**
+   * Tests the builder when customAttributes is a flat collection of key/value pairs using dotted-notation.
+   */
+  @Test
+  public void applyCustomSettingsWithFlatDottedNotation() {
+    final Map<String, Object> customSettings = this.customSettingsFlat();
+    final IncomingLinkSettings incomingLinksettings = IncomingLinkSettings.fromCustomSettings(customSettings).build();
+
+    assertThat(incomingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.JWT_HS_256);
+    assertThat(incomingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(incomingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(incomingLinksettings.encryptedTokenSharedSecret()).isEqualTo("incoming-credential");
+    assertThat(incomingLinksettings.getMinMessageWindow()).isEqualTo(Duration.ofMillis(2500));
+  }
+
+  /**
+   * Tests the builder when customAttributes is a Map of Maps.
+   */
+  @Test
+  public void applyCustomSettingsWithMapHeirarchy() {
+    final Map<String, Object> customSettings = this.customSettingsHeirarchical();
+    final IncomingLinkSettings incomingLinksettings = IncomingLinkSettings.fromCustomSettings(customSettings).build();
+
+    assertThat(incomingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.JWT_HS_256);
+    assertThat(incomingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(incomingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(incomingLinksettings.encryptedTokenSharedSecret()).isEqualTo("incoming-credential");
+    assertThat(incomingLinksettings.getMinMessageWindow()).isEqualTo(Duration.ofMillis(2500));
+  }
+
+  @Test
+  public void testWithoutCustomSettings() {
+    final IncomingLinkSettings incomingLinksettings =
+        IncomingLinkSettings.builder()
+            .authType(IlpOverHttpLinkSettings.AuthType.SIMPLE)
+            .tokenIssuer(HttpUrl.parse("https://incoming-issuer.example.com"))
+            .tokenAudience(HttpUrl.parse("https://incoming-audience.example.com/"))
+            .encryptedTokenSharedSecret(SHH)
+            .minMessageWindow(Duration.ofMillis(30))
+            .build();
+
+    assertThat(incomingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(incomingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-issuer.example.com/"));
+    assertThat(incomingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://incoming-audience.example.com/"));
+    assertThat(incomingLinksettings.encryptedTokenSharedSecret()).isEqualTo(SHH);
+    assertThat(incomingLinksettings.getMinMessageWindow()).isEqualTo(Duration.ofMillis(30));
+  }
+}
diff --git a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/OutgoingIlpOverIlpOverHttpLinkSettingsSettingsTest.java b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/OutgoingIlpOverIlpOverHttpLinkSettingsSettingsTest.java
new file mode 100644
index 0000000..3c1dbcf
--- /dev/null
+++ b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/OutgoingIlpOverIlpOverHttpLinkSettingsSettingsTest.java
@@ -0,0 +1,77 @@
+package org.interledger.link.http;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import okhttp3.HttpUrl;
+import org.junit.Test;
+
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Unit tests for {@link OutgoingLinkSettings}.
+ */
+public class OutgoingIlpOverIlpOverHttpLinkSettingsSettingsTest extends AbstractHttpLinkSettingsTest {
+
+  /**
+   * Tests the builder when customAttributes is a flat collection of key/value pairs using dotted-notation.
+   */
+  @Test
+  public void applyCustomSettingsWithFlatDottedNotation() {
+    final Map<String, Object> customSettings = this.customSettingsFlat();
+    final OutgoingLinkSettings outgoingLinksettings = OutgoingLinkSettings.fromCustomSettings(customSettings).build();
+
+    assertThat(outgoingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(outgoingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com/"));
+    assertThat(outgoingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com/"));
+    assertThat(outgoingLinksettings.tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(outgoingLinksettings.encryptedTokenSharedSecret()).isEqualTo("outgoing-credential");
+    assertThat(outgoingLinksettings.tokenExpiry().get()).isEqualTo(Duration.ofHours(24));
+    assertThat(outgoingLinksettings.url()).isEqualTo(HttpUrl.parse("https://outgoing.example.com/"));
+  }
+
+  /**
+   * Tests the builder when customAttributes is a Map of Maps.
+   */
+  @Test
+  public void applyCustomSettingsWithMapHeirarchy() {
+    final Map<String, Object> customSettings = this.customSettingsHeirarchical();
+    final OutgoingLinkSettings outgoingLinksettings = OutgoingLinkSettings.fromCustomSettings(customSettings).build();
+
+    assertThat(outgoingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(outgoingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com/"));
+    assertThat(outgoingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com/"));
+    assertThat(outgoingLinksettings.tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(outgoingLinksettings.encryptedTokenSharedSecret()).isEqualTo("outgoing-credential");
+    assertThat(outgoingLinksettings.tokenExpiry().get()).isEqualTo(Duration.ofHours(48));
+    assertThat(outgoingLinksettings.url()).isEqualTo(HttpUrl.parse("https://outgoing.example.com"));
+  }
+
+  @Test
+  public void testWithoutCustomSettings() {
+    final OutgoingLinkSettings outgoingLinksettings =
+        OutgoingLinkSettings.builder()
+            .authType(IlpOverHttpLinkSettings.AuthType.SIMPLE)
+            .tokenIssuer(HttpUrl.parse("https://outgoing-issuer.example.com"))
+            .tokenAudience(HttpUrl.parse("https://outgoing-audience.example.com"))
+            .tokenSubject("outgoing-subject")
+            .tokenExpiry(Duration.ofMillis(30))
+            .encryptedTokenSharedSecret("outgoing-credential")
+            .url(HttpUrl.parse("https://outgoing.example.com"))
+            .build();
+
+    assertThat(outgoingLinksettings.authType()).isEqualTo(IlpOverHttpLinkSettings.AuthType.SIMPLE);
+    assertThat(outgoingLinksettings.tokenIssuer().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-issuer.example.com"));
+    assertThat(outgoingLinksettings.tokenAudience().get())
+        .isEqualTo(HttpUrl.parse("https://outgoing-audience.example.com"));
+    assertThat(outgoingLinksettings.tokenSubject()).isEqualTo("outgoing-subject");
+    assertThat(outgoingLinksettings.tokenExpiry().get()).isEqualTo(Duration.ofMillis(30));
+    assertThat(outgoingLinksettings.encryptedTokenSharedSecret()).isEqualTo("outgoing-credential");
+    assertThat(outgoingLinksettings.url()).isEqualTo(HttpUrl.parse("https://outgoing.example.com"));
+  }
+}