class HgBufferProxy<T> {

    private final BlockingQueue<Supplier<T>> queue;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final ReentrantLock lock = new ReentrantLock();
    private final Runnable task;
    private Throwable err;

    private HgBufferProxy(Runnable task) {
        this.task = task;
        this.queue = new LinkedBlockingQueue<>();
    }

    public static HgBufferProxy of(Runnable task) {
        HgAssert.isArgumentNotNull(task, "task");
        return new HgBufferProxy(task);
    }

    public void send(T t) {
        if (t == null) {
            throw new IllegalArgumentException("the argument t is null");
        }
        if (this.closed.get()) {
            return;
        }
        this.lock.lock();
        try {
            this.queue.offer(() -> t);
        } finally {
            lock.unlock();
        }
    }

    private void apply() {
        this.lock.lock();
        try {
            if (!this.closed.get()) {
                this.task.run();
                Thread.yield();
            }
        } finally {
            this.lock.unlock();
        }
    }

    /**
     * return an item from the chan
     *
     * @return null when the chan has been closed
     * @throws RuntimeException
     */
    @CheckForNull
    public T receive(int time, Consumer<Integer> callback) {<FILL_FUNCTION_BODY>}

    public boolean isClosed() {
        return this.closed.get();
    }

    /**
     * @throws RuntimeException when fail to close the chan
     */
    public void close() {
        if (this.closed.get()) {
            return;
        }
        lock.lock();
        this.closed.set(true);
        try {
            this.queue.offer(() -> null);
        } finally {
            lock.unlock();
        }
    }

    public void setError(Throwable streamErr) {
        this.err = streamErr;
    }

    private void checkErr() {
        if (this.err != null) {
            throw HgStoreClientException.of(this.err);
        }
    }
}

        Supplier<T> s;
        if (this.closed.get()) {
            s = this.queue.poll();
            this.checkErr();
            return s != null ? s.get() : null;
        }
        if (this.queue.size() <= 1) {
            this.apply();
        }
        lock.lock();
        try {
            if (this.isClosed()) {
                s = this.queue.poll();
                this.checkErr();
                return s != null ? s.get() : null;
            }
        } finally {
            lock.unlock();
        }
        try {
            s = this.queue.poll(time, TimeUnit.SECONDS);
        } catch (Throwable t) {
            log.error("failed to receive a item from channel, cause by: ", t);
            throw HgStoreClientException.of(t);
        }
        if (s == null) {
            if (this.closed.get()) {
                s = this.queue.poll();
            } else {
                if (callback == null) {
                    throw new RuntimeException("timeout, max time: " + time + " seconds");
                } else {
                    callback.accept(time);
                }
            }
        }
        this.checkErr();
        return s != null ? s.get() : null;
    
<no_super_class>