shit.java

public static void main(String[] args) {
		OracleQueue<String> someQueue = new OracleQueue<>(50);

		ExecutorService es = Executors.newFixedThreadPool(4);

		Executors.newSingleThreadScheduledExecutor()
			.scheduleAtFixedRate(
				() -> System.out.println("Queue size:" + someQueue.queue.size()),
				1, 1, TimeUnit.SECONDS);

		Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(
			() -> someQueue.offer(UUID.randomUUID().toString()),
			1, 1, TimeUnit.MILLISECONDS
		);

		Observable.from(someQueue)
			.flatMap(el ->
				Observable.just(el)
					.subscribeOn(Schedulers.from(es))
					.doOnNext(s -> {
						System.out.println(s + " on " + Thread.currentThread().getName());
						try {
							Thread.sleep(1000);
							System.out.println("Done with " + s);
						} catch (Exception e) {
							;
						}
					})
			)
			.subscribe(Subscribers.empty());
	}


	private static class OracleQueue<T> implements Iterable<T> {
		private final BlockingQueue<T> queue;

		OracleQueue(int queueSize) {
			queue = new LinkedBlockingQueue<>(queueSize);
		}

		void offer(T element) {
			try {
				 queue.offer(element, 1, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
		}

		@Override
		public Iterator<T> iterator() {
			return new Iterator<T>() {
				@Override
				public boolean hasNext() {
					return true;
				}

				@Override
				public T next() {
					try {
						return queue.take();
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				}
			};
		}
	}