1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.util.thread.strategy;
20
21 import java.util.concurrent.Executor;
22
23 import org.eclipse.jetty.util.log.Log;
24 import org.eclipse.jetty.util.log.Logger;
25 import org.eclipse.jetty.util.thread.ExecutionStrategy;
26 import org.eclipse.jetty.util.thread.Locker;
27
28
29
30
31
32 public class ProduceConsume implements ExecutionStrategy, Runnable
33 {
34 private static final Logger LOG = Log.getLogger(ExecuteProduceConsume.class);
35
36 private final Locker _locker = new Locker();
37 private final Producer _producer;
38 private final Executor _executor;
39 private State _state = State.IDLE;
40
41 public ProduceConsume(Producer producer, Executor executor)
42 {
43 this._producer = producer;
44 this._executor = executor;
45 }
46
47 @Override
48 public void execute()
49 {
50 try (Locker.Lock lock = _locker.lock())
51 {
52 switch(_state)
53 {
54 case IDLE:
55 _state= State.PRODUCE;
56 break;
57
58 case PRODUCE:
59 case EXECUTE:
60 _state= State.EXECUTE;
61 return;
62 }
63 }
64
65
66 while (true)
67 {
68
69 Runnable task = _producer.produce();
70 if (LOG.isDebugEnabled())
71 LOG.debug("{} produced {}", _producer, task);
72
73 if (task == null)
74 {
75 try (Locker.Lock lock = _locker.lock())
76 {
77 switch(_state)
78 {
79 case IDLE:
80 throw new IllegalStateException();
81 case PRODUCE:
82 _state= State.IDLE;
83 return;
84 case EXECUTE:
85 _state= State.PRODUCE;
86 continue;
87 }
88 }
89 }
90
91
92 task.run();
93 }
94 }
95
96 @Override
97 public void dispatch()
98 {
99 _executor.execute(this);
100 }
101
102 @Override
103 public void run()
104 {
105 execute();
106 }
107
108 public static class Factory implements ExecutionStrategy.Factory
109 {
110 @Override
111 public ExecutionStrategy newExecutionStrategy(Producer producer, Executor executor)
112 {
113 return new ProduceConsume(producer, executor);
114 }
115 }
116
117 private enum State
118 {
119 IDLE, PRODUCE, EXECUTE
120 }
121 }