586      */

   586      */

   587     public void close() {

   587     public void close() {

   588         if (!closed) {

   588         if (!closed) {

   589             BufferedSubscription<T> b;

   589             BufferedSubscription<T> b;

   590             synchronized (this) {

   590             synchronized (this) {

   591                 b = clients;

   592                 b = clients;

   592                 clients = null;

   593                 clients = null;

   593                 closed = true;

   594                 closed = true;

   594             }

   595             }

   595             while (b != null) {

   596             while (b != null) {

   617             throw new NullPointerException();

   618             throw new NullPointerException();

   618         if (!closed) {

   619         if (!closed) {

   619             BufferedSubscription<T> b;

   620             BufferedSubscription<T> b;

   620             synchronized (this) {

   621             synchronized (this) {

   621                 b = clients;

   622                 b = clients;

   625             }

   628             }

   626             while (b != null) {

   629             while (b != null) {

   627                 BufferedSubscription<T> next = b.next;

   630                 BufferedSubscription<T> next = b.next;

   628                 b.next = null;

   631                 b.next = null;

   629                 b.onError(error);

   632                 b.onError(error);