Not logged in.  Login/Logout/Register | List snippets | | Create snippet | Upload image | Upload data

35
LINES

< > BotCompany Repo | #1031007 // moveProducerToNewThread

JavaX fragment (include) [tags: use-pretranspiled]

Transpiled version (3318L) is out of date.

static <A> Producer<A> moveProducerToNewThread(Producer<A> producer, CloseablesHolder ch, int bufferSize default 65536) {
  if (producer == null) null;
  SimpleCircularBuffer<A> buffer = new(bufferSize);
  new Flag done;
  new Var<Throwable> error;
  ch.add(() -> done.raise());
  startThread(r {
    try {
      A a;
      while ping (!done.isUp() && (a = producer.next()) != null) {
        ifdef moveProducerToNewThread_debug
          print("original producer returned: " + a);
        endifdef
        while (buffer.isFull() && !done.isUp()) sleep(1);
        buffer.add(a);
      }
    } catch e {
      error.set(e);
    } finally {
      done.raise();
    }
  });
  
  ret () -> {
    while true {
      bool isDone = done.isUp();
      A a = buffer.popFirst();
      if (a != null) ret a;
      if (isDone)
        null;
      else
        sleep(1);
    }
  };
}

download  show line numbers  debug dex  old transpilations   

Travelled to 4 computer(s): bhatertpkbcr, mqqgnosmbjvj, pyentgdyhuwx, vouqrxazstgt

No comments. add comment

Snippet ID: #1031007
Snippet name: moveProducerToNewThread
Eternal ID of this version: #1031007/6
Text MD5: 6a88618f5a0301b324d8026b27d51495
Author: stefan
Category: javax
Type: JavaX fragment (include)
Public (visible to everyone): Yes
Archived (hidden from active list): No
Created/modified: 2021-04-19 18:02:52
Source code size: 905 bytes / 35 lines
Pitched / IR pitched: No / No
Views / Downloads: 87 / 140
Version history: 5 change(s)
Referenced in: [show references]