Addresses multiple intertwining issues.

src/main/java/com/github/princesslana/eriscasper/ErisCasper.java

@@ -9,13 +9,15 @@
 import com.github.princesslana.eriscasper.rest.RouteCatalog;
 import com.github.princesslana.eriscasper.rest.Routes;
 import com.github.princesslana.eriscasper.util.OkHttp;
+import com.github.princesslana.eriscasper.util.Shard;
 import com.ufoscout.properlty.Properlty;
 import com.ufoscout.properlty.reader.EnvironmentVariablesReader;
 import com.ufoscout.properlty.reader.SystemPropertiesReader;
 import com.ufoscout.properlty.reader.decorator.ToLowerCaseAndDotKeyReader;
 import io.reactivex.Observable;
 import io.reactivex.Single;
 import io.reactivex.schedulers.Schedulers;
+import java.util.Optional;
 import okhttp3.OkHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,21 +40,21 @@ public class ErisCasper {
   private final Payloads payloads = new Payloads(jackson);
 
   private final Routes routes;
+  private final Optional<Shard> shard;
 
-  private ErisCasper(BotToken token) {
+  private ErisCasper(BotToken token, Optional<Shard> shard) {
     this.token = token;
-
+    this.shard = shard;
     routes = new Routes(token, httpClient, jackson);
   }
 
   private Observable<Event> getEvents() {
     Gateway gateway = Gateway.create(httpClient, payloads);
-
     return Single.just(RouteCatalog.getGateway())
         .observeOn(Schedulers.io())
         .flatMap(routes::execute)
         .toObservable()
-        .flatMap(gr -> gateway.connect(gr.getUrl(), token))
+        .flatMap(gr -> gateway.connect(gr.getUrl(), token, shard))
         .observeOn(Schedulers.computation())
         .share();
   }
@@ -75,6 +77,11 @@ public static ErisCasper create() {
   }
 
   public static ErisCasper create(String token) {
-    return new ErisCasper(BotToken.of(token));
+    return new ErisCasper(BotToken.of(token), Shard.fromConfig(CONFIG));
+  }
+
+  public static ErisCasper create(String token, int shardNumber, int shardTotal) {
+    Shard shard = new Shard(shardNumber, shardTotal);
+    return new ErisCasper(BotToken.of(token), Optional.of(shard));
   }
 }

src/main/java/com/github/princesslana/eriscasper/api/robot/Robot.java

@@ -34,7 +34,7 @@ public Completable apply(BotContext bctx) {
   }
 
   /**
-   * See {@link hear(Pattern, Function)}.
+   * See {@link #hear(Pattern, Function)}.
    *
    * @param regex regex to match
    * @param f function to execute when matched
@@ -55,7 +55,7 @@ public void hear(Pattern regex, Function<RobotContext, Completable> f) {
   }
 
   /**
-   * See {@link respond(Pattern, Function)}.
+   * See {@link #respond(Pattern, Function)}.
    *
    * @param regex regex to match
    * @param f function to execute when matched

src/main/java/com/github/princesslana/eriscasper/gateway/Gateway.java

@@ -1,12 +1,15 @@
 package com.github.princesslana.eriscasper.gateway;
 
 import com.github.princesslana.eriscasper.BotToken;
+import com.github.princesslana.eriscasper.ErisCasperFatalException;
 import com.github.princesslana.eriscasper.data.event.Event;
 import com.github.princesslana.eriscasper.data.event.HelloEventData;
 import com.github.princesslana.eriscasper.data.event.ReadyEvent;
+import com.github.princesslana.eriscasper.gateway.commands.ImmutableResume;
 import com.github.princesslana.eriscasper.rx.Singles;
 import com.github.princesslana.eriscasper.rx.websocket.RxWebSocket;
 import com.github.princesslana.eriscasper.rx.websocket.RxWebSocketEvent;
+import com.github.princesslana.eriscasper.util.Shard;
 import com.google.common.base.Preconditions;
 import io.github.resilience4j.ratelimiter.RateLimiter;
 import io.github.resilience4j.ratelimiter.RateLimiterConfig;
@@ -85,9 +88,21 @@ private void sequenceNumberSeen(Optional<SequenceNumber> seq) {
   }
 
   @SuppressWarnings("unchecked")
-  public Observable<Event> connect(String url, BotToken token) {
+  public Observable<Event> connect(String url, BotToken token, Optional<Shard> shard) {
+    Observable<RxWebSocketEvent> websocketEvents =
+        ws.connect(String.format("%s?v=%s&encoding=%s", url, VERSION, ENCODING));
+    websocketEvents
+        .ofType(RxWebSocketEvent.Closing.class)
+        .doOnNext(
+            e -> {
+              if (e.getCode() == 4004) {
+                e.getWebSocket().close(1002, "Invalid token.");
+                throw new ErisCasperFatalException("Failed to authenticate with discord servers.");
+              }
+            })
+        .share();
     Observable<Payload> ps =
-        ws.connect(String.format("%s?v=%s&encoding=%s", url, VERSION, ENCODING))
+        websocketEvents
             .ofType(RxWebSocketEvent.StringMessage.class)
             .map(RxWebSocketEvent.StringMessage::getText)
             .flatMapMaybe(
@@ -101,7 +116,8 @@ public Observable<Event> connect(String url, BotToken token) {
 
     Completable identify =
         ps.filter(Payload.isOp(OpCode.HELLO))
-            .flatMapCompletable(p -> isResumable() ? resume(ws, token) : identify(ws, token));
+            .flatMapCompletable(
+                p -> isResumable() ? resume(ws, token) : identify(ws, token, shard));
 
     Observable<Event> events = ps.flatMapMaybe(payloads::toEvent).share();
 
@@ -127,8 +143,8 @@ private Completable send(RxWebSocket ws, Payload payload) {
         .doOnComplete(() -> LOG.debug("Sent: {}.", payload));
   }
 
-  private Completable identify(RxWebSocket ws, BotToken token) {
-    return Single.just(payloads.identify(token))
+  private Completable identify(RxWebSocket ws, BotToken token, Optional<Shard> shard) {
+    return Single.just(payloads.identify(token, shard))
         .lift(RateLimiterOperator.of(identifyLimit))
         .flatMapCompletable(p -> send(ws, p));
   }

src/main/java/com/github/princesslana/eriscasper/gateway/Payloads.java

@@ -1,14 +1,16 @@
 package com.github.princesslana.eriscasper.gateway;
 
-import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.github.princesslana.eriscasper.BotToken;
 import com.github.princesslana.eriscasper.data.event.Event;
 import com.github.princesslana.eriscasper.data.event.EventFactory;
 import com.github.princesslana.eriscasper.data.immutable.Wrapped;
 import com.github.princesslana.eriscasper.data.immutable.Wrapper;
+import com.github.princesslana.eriscasper.gateway.commands.Identify;
+import com.github.princesslana.eriscasper.gateway.commands.ImmutableIdentify;
+import com.github.princesslana.eriscasper.gateway.commands.Resume;
 import com.github.princesslana.eriscasper.rx.Maybes;
+import com.github.princesslana.eriscasper.util.Shard;
 import io.reactivex.Maybe;
 import io.reactivex.Single;
 import java.util.Optional;
@@ -34,20 +36,20 @@ public Payload heartbeat(Optional<SequenceNumber> s) {
     return ImmutablePayload.builder().op(OpCode.HEARTBEAT).d(s.map(jackson::valueToTree)).build();
   }
 
-  public Payload identify(BotToken token) {
-    return identify(ImmutableIdentify.builder().token(token).build());
+  public Payload identify(BotToken token, Optional<Shard> shard) {
+    return identify(ImmutableIdentify.builder().token(token).shard(shard).build());
   }
 
   public Payload identify(Identify id) {
-    return ImmutablePayload.builder().op(OpCode.IDENTIFY).d(jackson.valueToTree(id)).build();
+    return id.toPayload(jackson);
   }
 
   public Single<Payload> read(String text) {
     return Single.fromCallable(() -> jackson.readValue(text, Payload.class));
   }
 
   public Payload resume(Resume r) {
-    return ImmutablePayload.builder().op(OpCode.RESUME).d(jackson.valueToTree(r)).build();
+    return r.toPayload(jackson);
   }
 
   public Maybe<Event> toEvent(Payload payload) {
@@ -63,61 +65,7 @@ public Single<String> writeToString(Payload p) {
     return Single.fromCallable(() -> jackson.writeValueAsString(p));
   }
 
-  /**
-   * @see <a href="https://discordapp.com/developers/docs/topics/gateway#identify">
-   *     https://discordapp.com/developers/docs/topics/gateway#identify</a>
-   */
-  // TODO: This structure is not complete
-  @Value.Immutable
-  @JsonDeserialize(as = ImmutableIdentify.class)
-  public static interface Identify {
-    BotToken getToken();
-
-    default ConnectionProperties getProperties() {
-      return ConnectionProperties.ofDefault();
-    }
-  }
-
-  /**
-   * @see <a
-   *     href="https://discordapp.com/developers/docs/topics/gateway#identify-identify-connection-properties">
-   *     https://discordapp.com/developers/docs/topics/gateway#identify-identify-connection-properties</a>
-   */
-  @Value.Immutable
-  public static interface ConnectionProperties {
-    @JsonProperty("$os")
-    String getOs();
-
-    @JsonProperty("$browser")
-    String getBrowser();
-
-    @JsonProperty("$device")
-    String getDevice();
-
-    public static ConnectionProperties ofDefault() {
-      return ImmutableConnectionProperties.builder()
-          .os(System.getProperty("os.name"))
-          .browser("ErisCasper.java")
-          .device("ErisCasper.java")
-          .build();
-    }
-  }
-
-  /**
-   * @see <a href="https://discordapp.com/developers/docs/topics/gateway#resume">
-   *     https://discordapp.com/developers/docs/topics/gateway#resume</a>
-   */
-  @Value.Immutable
-  public static interface Resume {
-    BotToken getToken();
-
-    @JsonProperty("session_id")
-    SessionId getSessionId();
-
-    SequenceNumber getSeq();
-  }
-
   @Value.Immutable
   @Wrapped
-  public static interface SessionIdWrapper extends Wrapper<String> {}
+  public interface SessionIdWrapper extends Wrapper<String> {}
 }

src/main/java/com/github/princesslana/eriscasper/gateway/commands/GatewayCommand.java

@@ -0,0 +1,8 @@
+package com.github.princesslana.eriscasper.gateway.commands;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.princesslana.eriscasper.gateway.Payload;
+
+public interface GatewayCommand {
+  Payload toPayload(ObjectMapper jackson);
+}

src/main/java/com/github/princesslana/eriscasper/gateway/commands/Identify.java

@@ -0,0 +1,60 @@
+package com.github.princesslana.eriscasper.gateway.commands;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.github.princesslana.eriscasper.BotToken;
+import com.github.princesslana.eriscasper.gateway.ImmutablePayload;
+import com.github.princesslana.eriscasper.gateway.OpCode;
+import com.github.princesslana.eriscasper.gateway.Payload;
+import com.github.princesslana.eriscasper.util.Shard;
+import java.util.Optional;
+import org.immutables.value.Value;
+
+/**
+ * @see <a href="https://discordapp.com/developers/docs/topics/gateway#identify">
+ *     https://discordapp.com/developers/docs/topics/gateway#identify</a>
+ */
+// TODO: This structure is not complete
+@JsonDeserialize(as = ImmutableIdentify.class)
+@Value.Immutable
+@Value.Enclosing
+public interface Identify extends GatewayCommand {
+  BotToken getToken();
+
+  Optional<Shard> shard();
+
+  default ConnectionProperties getProperties() {
+    return ConnectionProperties.ofDefault();
+  }
+
+  @Override
+  default Payload toPayload(ObjectMapper jackson) {
+    return ImmutablePayload.builder().op(OpCode.IDENTIFY).d(jackson.valueToTree(this)).build();
+  }
+
+  /**
+   * @see <a
+   *     href="https://discordapp.com/developers/docs/topics/gateway#identify-identify-connection-properties">
+   *     https://discordapp.com/developers/docs/topics/gateway#identify-identify-connection-properties</a>
+   */
+  @Value.Immutable
+  interface ConnectionProperties {
+    @JsonProperty("$os")
+    String getOs();
+
+    @JsonProperty("$browser")
+    String getBrowser();
+
+    @JsonProperty("$device")
+    String getDevice();
+
+    static ConnectionProperties ofDefault() {
+      return ImmutableIdentify.ConnectionProperties.builder()
+          .os(System.getProperty("os.name"))
+          .browser("ErisCasper.java")
+          .device("ErisCasper.java")
+          .build();
+    }
+  }
+}

src/main/java/com/github/princesslana/eriscasper/gateway/commands/Resume.java

@@ -0,0 +1,30 @@
+package com.github.princesslana.eriscasper.gateway.commands;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.princesslana.eriscasper.BotToken;
+import com.github.princesslana.eriscasper.gateway.ImmutablePayload;
+import com.github.princesslana.eriscasper.gateway.OpCode;
+import com.github.princesslana.eriscasper.gateway.Payload;
+import com.github.princesslana.eriscasper.gateway.SequenceNumber;
+import com.github.princesslana.eriscasper.gateway.SessionId;
+import org.immutables.value.Value;
+
+/**
+ * @see <a href="https://discordapp.com/developers/docs/topics/gateway#resume">
+ *     https://discordapp.com/developers/docs/topics/gateway#resume</a>
+ */
+@Value.Immutable
+public interface Resume extends GatewayCommand {
+  BotToken getToken();
+
+  @JsonProperty("session_id")
+  SessionId getSessionId();
+
+  SequenceNumber getSeq();
+
+  @Override
+  default Payload toPayload(ObjectMapper jackson) {
+    return ImmutablePayload.builder().op(OpCode.RESUME).d(jackson.valueToTree(this)).build();
+  }
+}

src/main/java/com/github/princesslana/eriscasper/rx/websocket/RxWebSocket.java

@@ -31,6 +31,7 @@ public Observable<RxWebSocketEvent> connect(String url) {
 
               ws = http.newWebSocket(rq, new Listener(em));
             })
+        .takeUntil(e -> e instanceof RxWebSocketEvent.Closed)
         .doOnNext(e -> LOG.trace("Received: {}.", e))
         .doOnError(e -> LOG.warn("Error: {}.", e));
   }
@@ -43,7 +44,7 @@ public Completable send(String text) {
   private static class Listener extends WebSocketListener {
     private final ObservableEmitter<RxWebSocketEvent> em;
 
-    public Listener(ObservableEmitter<RxWebSocketEvent> em) {
+    private Listener(ObservableEmitter<RxWebSocketEvent> em) {
       this.em = em;
     }