i++
	}
}

type mockReader struct {
	pos       int
	mockData  byte
	failAfter int
}

func newMockReader(failAfter int, data byte) *mockReader {
	return &mockReader{0, data, failAfter}
}

func (r *mockReader) Read(p []byte) (n int, err error) {
	if r.pos >= r.failAfter {
		return 0, fmt.Errorf("intentionally unexpected reader error")
	}
	r.pos++
	p[0] = r.mockData
	return 1, nil

	}
	return out
}

type fakeOutputWriter struct {
	b           bytes.Buffer
	injectError error
	failAfter   int
}

func (w *fakeOutputWriter) Write(p []byte) (n int, err error) {
	w.failAfter--
	if w.failAfter <= 0 && w.injectError != nil {
		return 0, w.injectError
	}
	return w.b.Write(p)
}
func (w *fakeOutputWriter) String() string { return w.b.String() }

	}
	if counter != 0 {
		t.Fatalf("expected 0 items, got %d", counter)
	}
}

func TestDecoderReaderFailure(t *testing.T) {
	var (
		failAfter = 900
		mockData  = byte('[')
	)

	r := newMockReader(failAfter, mockData)
	decoder := NewDecoder(r, -1)

	for mv := range decoder.Stream() {
		t.Logf("depth=%d offset=%d len=%d (%v)", mv.Depth, mv.Offset, mv.Length, mv.Value)
	}

   * even permanently blocked.
   */
  public void testListenerDeadlock() throws InterruptedException {
    final CountDownLatch failEnter = new CountDownLatch(1);
    final CountDownLatch failLeave = new CountDownLatch(1);
    final CountDownLatch afterStarted = new CountDownLatch(1);
    Service failRunService =
        new AbstractService() {