code: public void close() {
          if (log.isDebugEnabled()) {
              log.debug("close");
          }
          if (!subscriberStream.getState().equals(StreamState.CLOSED)) {
              IMessageInput in = msgInReference.get();
              if (in != null) {
                  in.unsubscribe(this);
                  msgInReference.set(null);
              }
              subscriberStream.setState(StreamState.CLOSED);
              clearWaitJobs();
              releasePendingMessage();
              lastMessageTs = 0;
              // XXX is clear ping required?
              //sendClearPing();
              InMemoryPushPushPipe out = (InMemoryPushPushPipe) msgOutReference.get();
              if (out != null) {
                  List<IConsumer> consumers = out.getConsumers();
                  // assume a list of 1 in most cases
                  if (log.isDebugEnabled()) {
                      log.debug("Message out consumers: {}", consumers.size());
                  }
                  if (!consumers.isEmpty()) {
                      for (IConsumer consumer : consumers) {
                          out.unsubscribe(consumer);
                      }
                  }
                  msgOutReference.set(null);
              }
          } else {
              log.debug("Stream is already in closed state");
          }
      }
docs: Close stream