hunt.concurrency.CountingCallback source code

1 /*
2  * Hunt - A refined core library for D programming language.
3  *
4  * Copyright (C) 2018-2019 HuntLabs
5  *
6  * Website: https://www.huntlabs.net/
7  *
8  * Licensed under the Apache-2.0 License.
9  *
10  */
11 
12 module hunt.concurrency.CountingCallback;
13 
14 import hunt.util.Common;
15 import core.atomic;
16 import std.format;
17 
18 
19 /**
20  * <p>
21  * A callback wrapper that succeeds the wrapped callback when the count is
22  * reached, or on first failure.
23  * </p>
24  * <p>
25  * This callback is particularly useful when an async operation is split into
26  * multiple parts, for example when an original byte buffer that needs to be
27  * written, along with a callback, is split into multiple byte buffers, since it
28  * allows the original callback to be wrapped and notified only when the last
29  * part has been processed.
30  * </p>
31  * <p>
32  * Example:
33  * </p>
34  * 
35  * <pre>
36  * void process(EndPoint endPoint, ByteBuffer buffer, Callback callback) {
37  * 	ByteBuffer[] buffers = split(buffer);
38  * 	CountCallback countCallback = new CountCallback(callback, buffers.length);
39  * 	endPoint.write(countCallback, buffers);
40  * }
41  * </pre>
42  */
43 class CountingCallback : NestedCallback {
44 	private shared(int) count;
45 
46 	this(Callback callback, int count) {
47 		super(callback);
48 		this.count = count;
49 	}
50 
51 	override
52 	void succeeded() {
53 		// Forward success on the last success.
54 		while (true) {
55 			int current = count;
56 
57 			// Already completed ?
58 			if (current == 0)
59 				return;
60 
61 			// if (count.compareAndSet(current, current - 1)) 
62             if(count == current)
63             {
64                 count = current - 1;
65 				if (current == 1)
66 					super.succeeded();
67 				return;
68 			}
69 		}
70 	}
71 
72 	override
73 	void failed(Exception failure) {
74 		// Forward failure on the first failure.
75 		while (true) {
76 			int current = count;
77 
78 			// Already completed ?
79 			if (current == 0)
80 				return;
81 
82 			// if (count.compareAndSet(current, 0)) 
83              if(count == current)
84             {
85                 count = 0;
86 				super.failed(failure);
87 				return;
88 			}
89 		}
90 	}
91 
92 	override
93 	string toString() {
94 		return format("%s@%d", typeof(this).stringof, toHash());
95 	}
96 }