<commit_before>/*
 * Copyright (C) 2017 <NAME> (殷啟聰)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */

package chat.viska.xmpp;

import chat.viska.commons.ExceptionCaughtEvent;
import chat.viska.commons.reactive.MutableReactiveObject;
import chat.viska.commons.reactive.ReactiveObject;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.Schedulers;
import java.util.Collections;
import java.util.EventObject;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.WillCloseWhenClosed;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.lang3.StringUtils;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

/**
 * XMPP session defining abstract concept for all implementations.
 */
@ThreadSafe
public abstract class Session implements AutoCloseable {

  /**
   * State of a {@link Session}.
   *
   * <h1>State diagram</h1>
   * <pre>{@code
   *             +--------------+
   *             |              |
   *             |   DISPOSED   |
   *             |              |
   *             +--------------+
   *
   *                    ^
   *                    | dispose()
   *                    +
   *
   *             +---------------+
   *             |               |                            Connection loss
   * +---------> | DISCONNECTED  |    <------------+------------------------------+-----------------------+
   * |           |               |                 |                              |                       |
   * |           +---------------+                 |                              |                       |
   * |                                             |                              |                       |
   * |                 +  ^                        |                              |                       |
   * |         login() |  | Connection loss        |                              |                       |
   * |                 v  +                        +                              +                       +
   * |
   * |           +--------------+           +--------------+               +--------------+        +--------------+
   * |           |              |           |              |               |              |        |              |
   * |           |  CONNECTING  | +-------> |  CONNECTED   | +---------->  | HANDSHAKING  | +----> |   ONLINE     |
   * |           |              |           |              |               |              |        |              |
   * |           +--------------+           +--------------+               +--------------+        +--------------+
   * |
   * |                  +                           +                             +                       +
   * |                  | disconnect()              |                             |                       |
   * |                  v                           |                             |                       |
   * |                                              |                             |                       |
   * |           +---------------+                  |                             |                       |
   * |           |               |                  |                             |                       |
   * +---------+ | DISCONNECTING | <----------------+-----------------------------+-----------------------+
   *             |               |                             disconnect()
   *             +---------------+
   * }</pre>
   */
  public enum State {

    /**
     * Indicates a network connection to the server is established and is
     * about to login or perform in-band registration.
     */
    CONNECTED,

    /**
     * Indicates the {@link Session} is establishing a network
     * connection to a server.
     */
    CONNECTING,

    /**
     * Indicates the network connection is lost and waiting to reconnect and
     * resume the XMPP stream. However, it enters {@link State#DISPOSED} directly
     * upon losing the connection if
     * <a href="https://xmpp.org/extensions/xep-0198.html">Stream
     * Management</a> is disabled.
     */
    DISCONNECTED,

    /**
     * Indicates the {@link Session} is closing the connection or the
     * server is doing so.
     */
    DISCONNECTING,

    /**
     * Indicates the {@link Session} has been shut down. Most actions
     * that changes the state will throw an {@link IllegalStateException}.
     */
    DISPOSED,

    /**
     * Indicates the {@link Session} is logging in.
     */
    HANDSHAKING,

    /**
     * Indicates the user has logged into the server.
     */
    ONLINE
  }

  /**
   * Contains information of {@link Plugin}s applied on an {@link Session}.
   */
  @ThreadSafe
  public class PluginManager implements SessionAware {

    @GuardedBy("itself")
    private final Set<PluginContext> contexts = new HashSet<>();

    private PluginManager() {}

    /**
     * Applies a {@link Plugin}. This method does nothing if the plugin
     * has already been applied.
     * @throws IllegalArgumentException If it fails to apply the {@link Plugin}.
     */
    public void apply(@Nonnull final Class<? extends Plugin> type) {
      if (getPlugin(type) != null) {
        return;
      }
      synchronized (this.contexts) {
        if (getPlugin(type) != null) {
          return;
        }
        final Plugin plugin;
        try {
          plugin = type.getConstructor().newInstance();
        } catch (Exception ex) {
          throw new IllegalArgumentException(
              ex
          );
        }
        for (Class<? extends Plugin> it : plugin.getDependencies()) {
          apply(it);
        }
        final PluginContext context = new PluginContext(plugin);
        this.contexts.add(context);
        plugin.onApplying(context);
      }
    }

    /**
     * Gets an applied plugin which is of a particular type.
     * @return {@code null} if the plugin cannot be found.
     */
    @Nullable
    public <T extends Plugin> T getPlugin(@Nonnull final Class<T> type) {
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            return (T) it.plugin;
          }
        }
      }
      return null;
    }

    @Nonnull
    public Set<Plugin> getPlugins() {
      final Set<Plugin> results = new HashSet<>();
      Observable.fromIterable(this.contexts).map(it -> it.plugin).subscribe(results::add);
      return Collections.unmodifiableSet(results);
    }

    public void setEnabled(@Nonnull final Class<? extends Plugin> type, final boolean enabled) {
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            it.enabled.changeValue(false);
          }
        }
      }
    }

    public void remove(@Nonnull final Class<? extends Plugin> type) {
      final Set<PluginContext> toRemove = new HashSet<>();
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            it.onDestroying();
            toRemove.add(it);
          }
        }
        this.contexts.removeAll(toRemove);
      }
    }

    public Set<String> getAllFeatures() {
      final Set<String> features = new LinkedHashSet<>();
      synchronized (contexts) {
        for (PluginContext it : contexts) {
          features.addAll(it.plugin.getFeatures());
        }
      }
      return features;
    }

    @Override
    @Nonnull
    public Session getSession() {
      return Session.this;
    }
  }

  /**
   * Represents a processing window offered by a {@link Session} to a {@link Plugin}. When the
   * {@link Plugin} is disabled or the {@link Session} is not online, the context will become
   * unavailable during which no {@link Stanza} may be sent or received. The context will however
   * buffer all pending outbound {@link Stanza}s and send them all once it becomes available. When
   * the {@link Plugin} is removed from the {@link Session}, the context is destroyed and all its
   * stream properties will signal a completion.
   */
  public class PluginContext implements SessionAware {

    @GuardedBy("itself")
    private final MutableReactiveObject<Boolean> enabled = new MutableReactiveObject<>(true);
    private final MutableReactiveObject<Boolean> available = new MutableReactiveObject<>(
        enabled.getValue() && getState().getValue() == State.ONLINE
    );
    private final Plugin plugin;
    private final FlowableProcessor<Stanza> inboundStanzaStream = PublishProcessor.create();
    private final Disposable stanzaSubscription;

    private PluginContext(@Nonnull final Plugin plugin) {
      this.plugin = plugin;
      Flowable.combineLatest(
          this.enabled.getStream(),
          getState().getStream(),
          (enabled, state) -> enabled && state == State.ONLINE
      ).subscribe(this.available::changeValue);
      this.stanzaSubscription = Session.this.getInboundStanzaStream().filter(
          it -> this.enabled.getValue()
      ).subscribe(
          this.inboundStanzaStream::onNext
      );
      getState().getStream().filter(it -> it == State.DISPOSED).firstOrError().subscribe( it -> {
        onDestroying();
      });
    }

    private void onDestroying() {
      this.stanzaSubscription.dispose();
      this.inboundStanzaStream.onComplete();
      synchronized (this.enabled) {
        this.enabled.complete();
      }
      this.available.complete();
    }

    /**
     * Sends a stanza.
     */
    @Nonnull
    public StanzaReceipt sendStanza(@Nonnull final Stanza stanza) {
      this.available.getStream().filter(it -> it).firstElement().subscribe(
          it -> Session.this.sendStanza(stanza)
      );
      return new StanzaReceipt(Session.this, Maybe.empty());
    }

    /**
     * Sends an {@code <iq/>}.
     * @return {@link IqReceipt} whose {@link IqReceipt#getResponse()} will signal a completion
     *         immediately if the provided {@code iq} is a result or error.
     */
    @Nonnull
    public IqReceipt sendIq(@Nonnull final Stanza iq) {
      if (StringUtils.isBlank(iq.getId())) {
        throw new IllegalArgumentException("No ID");
      }
      final Maybe<Stanza> response;
      if (iq.getType() != Stanza.Type.IQ) {
        throw new IllegalArgumentException("Not an <iq/>.");
      } else if (iq.getIqType() == Stanza.IqType.GET || iq.getIqType() == Stanza.IqType.SET) {
        response = getInboundStanzaStream().filter(
            it -> iq.getId().equals(it.getId())
        ).firstElement().doOnSuccess(it -> {
          if (it.getIqType() == Stanza.IqType.ERROR) {
            final StanzaErrorException error;
            try {
              error = StanzaErrorException.fromXml(it.getXml());
              throw error;
            } catch (StreamErrorException ex) {
              sendError(ex);
            }
          }
        });
      } else if (iq.getIqType() == null) {
        throw new IllegalArgumentException("<iq/> has no type.");
      } else {
        response = Maybe.empty();
      }
      return new IqReceipt(Session.this, sendStanza(iq).getServerAcknowledment(), response);
    }

    /**
     * Sends a simple query. This query will look like:
     * <p>{@code <iq id="..." to="target"><query xmlns="namespace" (more params) />}</p>
     */
    @Nonnull
    public IqReceipt sendIq(@Nonnull final String namespace,
                            @Nullable final Jid recipient,
                            @Nullable final Map<String, String> attributes) {
      final String id = UUID.randomUUID().toString();
      final Document iq =  Stanza.getIqTemplate(
          Stanza.IqType.GET,
          id,
          getNegotiatedJid(),
          recipient
      );
      final Element element = (Element) iq.getDocumentElement().appendChild(
          iq.createElementNS(namespace, "query")
      );
      if (attributes != null) {
        for (Map.Entry<String, String> it : attributes.entrySet()) {
          element.setAttribute(it.getKey(), it.getValue());
        }
      }
      return sendIq(new XmlWrapperStanza(iq));
    }

    /**
     * Sends a stanza error.
     */
    @Nonnull
    public StanzaReceipt sendError(@Nonnull final StanzaErrorException error) {
      return sendStanza(new XmlWrapperStanza(error.toXml()));
    }

    public void sendError(@Nonnull final StreamErrorException error) {
      isAvailable().getStream().filter(it -> it).firstElement().subscribe(
          it -> Session.this.sendError(error)
      );
    }

    /**
     * Gets a stream of inbound {@link Stanza}s.
     */
    @Nonnull
    public Flowable<Stanza> getInboundStanzaStream() {
      return inboundStanzaStream;
    }

    public ReactiveObject<Boolean> isAvailable() {
      return available;
    }

    @Nonnull
    @Override
    public Session getSession() {
      return Session.this;
    }
  }

  private final MutableReactiveObject<State> state = new MutableReactiveObject<>(State.DISCONNECTED);
  private final FlowableProcessor<EventObject> eventStream;
  private final PluginManager pluginManager = new PluginManager();
  private final Logger logger = Logger.getAnonymousLogger();
  private Jid loginJid = Jid.EMPTY;
  private Jid authzId = Jid.EMPTY;
  private boolean neverOnline = true;

  private void onDisposing() {
    this.eventStream.onComplete();
    this.state.complete();
  }

  protected void log(final EventObject event) {
    logger.log(Level.FINE, event.toString());
  }

  protected void log(final ExceptionCaughtEvent event) {
    logger.log(Level.SEVERE, event.toString(), event.getCause());
  }

  protected Session() {
    // Event stream
    final FlowableProcessor<EventObject> unsafeEventStream = PublishProcessor.create();
    this.eventStream = unsafeEventStream.toSerialized();
    getState().getStream().filter(it -> it == State.ONLINE).firstElement().subscribe(
        it -> neverOnline = false
    );
    getState().getStream().filter(
        it -> it == State.DISPOSED
    ).firstOrError().observeOn(Schedulers.io()).subscribe(<commit_msg>Session: Logs events at INFO level<commit_after>/*
 * Copyright (C) 2017 <NAME> (殷啟聰)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */

package chat.viska.xmpp;

import chat.viska.commons.ExceptionCaughtEvent;
import chat.viska.commons.reactive.MutableReactiveObject;
import chat.viska.commons.reactive.ReactiveObject;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.Schedulers;
import java.util.Collections;
import java.util.EventObject;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.WillCloseWhenClosed;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.lang3.StringUtils;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

/**
 * XMPP session defining abstract concept for all implementations.
 */
@ThreadSafe
public abstract class Session implements AutoCloseable {

  /**
   * State of a {@link Session}.
   *
   * <h1>State diagram</h1>
   * <pre>{@code
   *             +--------------+
   *             |              |
   *             |   DISPOSED   |
   *             |              |
   *             +--------------+
   *
   *                    ^
   *                    | dispose()
   *                    +
   *
   *             +---------------+
   *             |               |                            Connection loss
   * +---------> | DISCONNECTED  |    <------------+------------------------------+-----------------------+
   * |           |               |                 |                              |                       |
   * |           +---------------+                 |                              |                       |
   * |                                             |                              |                       |
   * |                 +  ^                        |                              |                       |
   * |         login() |  | Connection loss        |                              |                       |
   * |                 v  +                        +                              +                       +
   * |
   * |           +--------------+           +--------------+               +--------------+        +--------------+
   * |           |              |           |              |               |              |        |              |
   * |           |  CONNECTING  | +-------> |  CONNECTED   | +---------->  | HANDSHAKING  | +----> |   ONLINE     |
   * |           |              |           |              |               |              |        |              |
   * |           +--------------+           +--------------+               +--------------+        +--------------+
   * |
   * |                  +                           +                             +                       +
   * |                  | disconnect()              |                             |                       |
   * |                  v                           |                             |                       |
   * |                                              |                             |                       |
   * |           +---------------+                  |                             |                       |
   * |           |               |                  |                             |                       |
   * +---------+ | DISCONNECTING | <----------------+-----------------------------+-----------------------+
   *             |               |                             disconnect()
   *             +---------------+
   * }</pre>
   */
  public enum State {

    /**
     * Indicates a network connection to the server is established and is
     * about to login or perform in-band registration.
     */
    CONNECTED,

    /**
     * Indicates the {@link Session} is establishing a network
     * connection to a server.
     */
    CONNECTING,

    /**
     * Indicates the network connection is lost and waiting to reconnect and
     * resume the XMPP stream. However, it enters {@link State#DISPOSED} directly
     * upon losing the connection if
     * <a href="https://xmpp.org/extensions/xep-0198.html">Stream
     * Management</a> is disabled.
     */
    DISCONNECTED,

    /**
     * Indicates the {@link Session} is closing the connection or the
     * server is doing so.
     */
    DISCONNECTING,

    /**
     * Indicates the {@link Session} has been shut down. Most actions
     * that changes the state will throw an {@link IllegalStateException}.
     */
    DISPOSED,

    /**
     * Indicates the {@link Session} is logging in.
     */
    HANDSHAKING,

    /**
     * Indicates the user has logged into the server.
     */
    ONLINE
  }

  /**
   * Contains information of {@link Plugin}s applied on an {@link Session}.
   */
  @ThreadSafe
  public class PluginManager implements SessionAware {

    @GuardedBy("itself")
    private final Set<PluginContext> contexts = new HashSet<>();

    private PluginManager() {}

    /**
     * Applies a {@link Plugin}. This method does nothing if the plugin
     * has already been applied.
     * @throws IllegalArgumentException If it fails to apply the {@link Plugin}.
     */
    public void apply(@Nonnull final Class<? extends Plugin> type) {
      if (getPlugin(type) != null) {
        return;
      }
      synchronized (this.contexts) {
        if (getPlugin(type) != null) {
          return;
        }
        final Plugin plugin;
        try {
          plugin = type.getConstructor().newInstance();
        } catch (Exception ex) {
          throw new IllegalArgumentException(
              ex
          );
        }
        for (Class<? extends Plugin> it : plugin.getDependencies()) {
          apply(it);
        }
        final PluginContext context = new PluginContext(plugin);
        this.contexts.add(context);
        plugin.onApplying(context);
      }
    }

    /**
     * Gets an applied plugin which is of a particular type.
     * @return {@code null} if the plugin cannot be found.
     */
    @Nullable
    public <T extends Plugin> T getPlugin(@Nonnull final Class<T> type) {
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            return (T) it.plugin;
          }
        }
      }
      return null;
    }

    @Nonnull
    public Set<Plugin> getPlugins() {
      final Set<Plugin> results = new HashSet<>();
      Observable.fromIterable(this.contexts).map(it -> it.plugin).subscribe(results::add);
      return Collections.unmodifiableSet(results);
    }

    public void setEnabled(@Nonnull final Class<? extends Plugin> type, final boolean enabled) {
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            it.enabled.changeValue(false);
          }
        }
      }
    }

    public void remove(@Nonnull final Class<? extends Plugin> type) {
      final Set<PluginContext> toRemove = new HashSet<>();
      synchronized (this.contexts) {
        for (PluginContext it : this.contexts) {
          if (type.isInstance(it.plugin)) {
            it.onDestroying();
            toRemove.add(it);
          }
        }
        this.contexts.removeAll(toRemove);
      }
    }

    public Set<String> getAllFeatures() {
      final Set<String> features = new LinkedHashSet<>();
      synchronized (contexts) {
        for (PluginContext it : contexts) {
          features.addAll(it.plugin.getFeatures());
        }
      }
      return features;
    }

    @Override
    @Nonnull
    public Session getSession() {
      return Session.this;
    }
  }

  /**
   * Represents a processing window offered by a {@link Session} to a {@link Plugin}. When the
   * {@link Plugin} is disabled or the {@link Session} is not online, the context will become
   * unavailable during which no {@link Stanza} may be sent or received. The context will however
   * buffer all pending outbound {@link Stanza}s and send them all once it becomes available. When
   * the {@link Plugin} is removed from the {@link Session}, the context is destroyed and all its
   * stream properties will signal a completion.
   */
  public class PluginContext implements SessionAware {

    @GuardedBy("itself")
    private final MutableReactiveObject<Boolean> enabled = new MutableReactiveObject<>(true);
    private final MutableReactiveObject<Boolean> available = new MutableReactiveObject<>(
        enabled.getValue() && getState().getValue() == State.ONLINE
    );
    private final Plugin plugin;
    private final FlowableProcessor<Stanza> inboundStanzaStream = PublishProcessor.create();
    private final Disposable stanzaSubscription;

    private PluginContext(@Nonnull final Plugin plugin) {
      this.plugin = plugin;
      Flowable.combineLatest(
          this.enabled.getStream(),
          getState().getStream(),
          (enabled, state) -> enabled && state == State.ONLINE
      ).subscribe(this.available::changeValue);
      this.stanzaSubscription = Session.this.getInboundStanzaStream().filter(
          it -> this.enabled.getValue()
      ).subscribe(
          this.inboundStanzaStream::onNext
      );
      getState().getStream().filter(it -> it == State.DISPOSED).firstOrError().subscribe( it -> {
        onDestroying();
      });
    }

    private void onDestroying() {
      this.stanzaSubscription.dispose();
      this.inboundStanzaStream.onComplete();
      synchronized (this.enabled) {
        this.enabled.complete();
      }
      this.available.complete();
    }

    /**
     * Sends a stanza.
     */
    @Nonnull
    public StanzaReceipt sendStanza(@Nonnull final Stanza stanza) {
      this.available.getStream().filter(it -> it).firstElement().subscribe(
          it -> Session.this.sendStanza(stanza)
      );
      return new StanzaReceipt(Session.this, Maybe.empty());
    }

    /**
     * Sends an {@code <iq/>}.
     * @return {@link IqReceipt} whose {@link IqReceipt#getResponse()} will signal a completion
     *         immediately if the provided {@code iq} is a result or error.
     */
    @Nonnull
    public IqReceipt sendIq(@Nonnull final Stanza iq) {
      if (StringUtils.isBlank(iq.getId())) {
        throw new IllegalArgumentException("No ID");
      }
      final Maybe<Stanza> response;
      if (iq.getType() != Stanza.Type.IQ) {
        throw new IllegalArgumentException("Not an <iq/>.");
      } else if (iq.getIqType() == Stanza.IqType.GET || iq.getIqType() == Stanza.IqType.SET) {
        response = getInboundStanzaStream().filter(
            it -> iq.getId().equals(it.getId())
        ).firstElement().doOnSuccess(it -> {
          if (it.getIqType() == Stanza.IqType.ERROR) {
            final StanzaErrorException error;
            try {
              error = StanzaErrorException.fromXml(it.getXml());
              throw error;
            } catch (StreamErrorException ex) {
              sendError(ex);
            }
          }
        });
      } else if (iq.getIqType() == null) {
        throw new IllegalArgumentException("<iq/> has no type.");
      } else {
        response = Maybe.empty();
      }
      return new IqReceipt(Session.this, sendStanza(iq).getServerAcknowledment(), response);
    }

    /**
     * Sends a simple query. This query will look like:
     * <p>{@code <iq id="..." to="target"><query xmlns="namespace" (more params) />}</p>
     */
    @Nonnull
    public IqReceipt sendIq(@Nonnull final String namespace,
                            @Nullable final Jid recipient,
                            @Nullable final Map<String, String> attributes) {
      final String id = UUID.randomUUID().toString();
      final Document iq =  Stanza.getIqTemplate(
          Stanza.IqType.GET,
          id,
          getNegotiatedJid(),
          recipient
      );
      final Element element = (Element) iq.getDocumentElement().appendChild(
          iq.createElementNS(namespace, "query")
      );
      if (attributes != null) {
        for (Map.Entry<String, String> it : attributes.entrySet()) {
          element.setAttribute(it.getKey(), it.getValue());
        }
      }
      return sendIq(new XmlWrapperStanza(iq));
    }

    /**
     * Sends a stanza error.
     */
    @Nonnull
    public StanzaReceipt sendError(@Nonnull final StanzaErrorException error) {
      return sendStanza(new XmlWrapperStanza(error.toXml()));
    }

    public void sendError(@Nonnull final StreamErrorException error) {
      isAvailable().getStream().filter(it -> it).firstElement().subscribe(
          it -> Session.this.sendError(error)
      );
    }

    /**
     * Gets a stream of inbound {@link Stanza}s.
     */
    @Nonnull
    public Flowable<Stanza> getInboundStanzaStream() {
      return inboundStanzaStream;
    }

    public ReactiveObject<Boolean> isAvailable() {
      return available;
    }

    @Nonnull
    @Override
    public Session getSession() {
      return Session.this;
    }
  }

  private final MutableReactiveObject<State> state = new MutableReactiveObject<>(State.DISCONNECTED);
  private final FlowableProcessor<EventObject> eventStream;
  private final PluginManager pluginManager = new PluginManager();
  private final Logger logger = Logger.getAnonymousLogger();
  private Jid loginJid = Jid.EMPTY;
  private Jid authzId = Jid.EMPTY;
  private boolean neverOnline = true;

  private void onDisposing() {
    this.eventStream.onComplete();
    this.state.complete();
  }

  protected void log(final EventObject event) {
    logger.log(Level.INFO, event.toString());
  }

  protected void log(final ExceptionCaughtEvent event) {
    logger.log(Level.SEVERE, event.toString(), event.getCause());
  }

  protected Session() {
    // Event stream
    final FlowableProcessor<EventObject> unsafeEventStream = PublishProcessor.create();
    this.eventStream = unsafeEventStream.toSerialized();
    getState().getStream().filter(it -> it == State.ONLINE).firstElement().subscribe(
        it -> neverOnline = false
    );
    getState().getStream().filter(
        it -> it == State.DISPOSED
    ).firstOrError().observeOn(Schedulers.io()).subscribe(
{'fraction_non_alphanumeric': 0.12967786606877907, 'fraction_numerical': 0.0011210392693461703, 'mean_word_length': 1.6656705923712427, 'pattern_counts': {'":': 0, '<': 95, '<?xml version=': 0, '>': 127, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 112, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2405898', 'n_tokens_mistral': 8075, 'n_tokens_neox': 7855, 'n_words': 1952}