#353

Committed 21 Jul 2025 10:01PM UTC coverage: 71.342% (+1.2%) from 70.173%

Build # #353

Build Type

push

Committed by

Commit Message

Implement ZeroMQ instead raw socket. Add and fix some tests.

Run Details

43 of 47 new or added lines in 4 files covered. (91.49%)

4 existing lines in 2 files now uncovered.

1536 of 2153 relevant lines covered (71.34%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.39

/src/main/java/net/ladenthin/bitcoinaddressfinder/Finder.java

// @formatter:off
/**
 * Copyright 2020 Bernard Ladenthin bernard.ladenthin@gmail.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
// @formatter:on
package net.ladenthin.bitcoinaddressfinder;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import net.ladenthin.bitcoinaddressfinder.configuration.CFinder;
import net.ladenthin.bitcoinaddressfinder.configuration.CProducer;
import net.ladenthin.bitcoinaddressfinder.persistence.PersistenceUtils;
import org.bitcoinj.base.Network;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.function.*;

public class Finder implements Interruptable {

    /**
     * We must define a maximum time to wait for terminate. Wait for 100 thousand years is enough.
     */
    @VisibleForTesting
    static Duration AWAIT_DURATION_TERMINATE = Duration.ofDays(365L * 1000L);
    
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    
    private final CFinder finder;
    
    private final Map<String, KeyProducer> keyProducers = new HashMap<>();
    
    @Nullable
    private ConsumerJava consumerJava;
    
    private final List<ProducerOpenCL> openCLProducers = new ArrayList<>();
    private final List<ProducerJava> javaProducers = new ArrayList<>();
    private final List<ProducerJavaSecretsFiles> javaProducersSecretsFiles = new ArrayList<>();
    
    @VisibleForTesting
    final ExecutorService producerExecutorService = Executors.newCachedThreadPool();
    
    private final Network network = new NetworkParameterFactory().getNetwork();
    private final KeyUtility keyUtility = new KeyUtility(network, new ByteBufferUtility(false));
    private final PersistenceUtils persistenceUtils = new PersistenceUtils(network);
    private final BitHelper bitHelper = new BitHelper();
    
    public Finder(CFinder finder) {
        this.finder = finder;
    }
    
    public void startKeyProducer() {
        logger.info("startKeyProducer");
        processKeyProducers(
            finder.keyProducerJavaRandom,
            cKeyProducerJavaRandom -> new KeyProducerJavaRandom(cKeyProducerJavaRandom, keyUtility, bitHelper),
            cKeyProducerJavaRandom -> cKeyProducerJavaRandom.keyProducerId,
            keyProducers
        );

        processKeyProducers(
            finder.keyProducerJavaBip39,
            cKeyProducerJavaBip39 -> new KeyProducerJavaBip39(cKeyProducerJavaBip39, keyUtility, bitHelper),
            cKeyProducerJavaBip39 -> cKeyProducerJavaBip39.keyProducerId,
            keyProducers
        );

        processKeyProducers(
            finder.keyProducerJavaIncremental,
            cKeyProducerJavaIncremental -> new KeyProducerJavaIncremental(cKeyProducerJavaIncremental, keyUtility, bitHelper),
            cKeyProducerJavaIncremental -> cKeyProducerJavaIncremental.keyProducerId,
            keyProducers
        );

        processKeyProducers(
            finder.keyProducerJavaZmq,
            cKeyProducerJavaZmq -> new KeyProducerJavaZmq(cKeyProducerJavaZmq, keyUtility, bitHelper),
            cKeyProducerJavaZmq -> cKeyProducerJavaZmq.keyProducerId,
            keyProducers
        );
    }
    
    private <T, K> void processKeyProducers(
        List<T> configList,
        Function<T, K> constructor,
        Function<T, String> getId,
        Map<String, K> keyProducers
    ) {
        if (configList != null) {
            for (T config : configList) {
                String keyProducerId = getId.apply(config);
                if (keyProducerId == null) {
                    throw new KeyProducerIdNullException();
                }
                if (keyProducers.containsKey(keyProducerId)) {
                    throw new KeyProducerIdIsNotUniqueException(keyProducerId);
                }
                K keyProducer = constructor.apply(config);
                keyProducers.put(keyProducerId, keyProducer);
            }
        }
    }

    public void startConsumer() {
        logger.info("startConsumer");
        if (finder.consumerJava != null) {
            consumerJava = new ConsumerJava(finder.consumerJava, keyUtility, persistenceUtils);
            consumerJava.initLMDB();
            consumerJava.startConsumer();
            consumerJava.startStatisticsTimer();
        }
    }

    public void configureProducer() {
        logger.info("configureProducer");
        processProducers(
            finder.producerJava,
            bitHelper::assertBatchSizeInBitsIsInRange,
            this::getKeyProducer,
            (config, keyProducer) -> new ProducerJava(config, consumerJava, keyUtility, keyProducer, bitHelper),
            javaProducers
        );

        processProducers(
            finder.producerJavaSecretsFiles,
            bitHelper::assertBatchSizeInBitsIsInRange,
            this::getKeyProducer,
            (config, keyProducer) -> new ProducerJavaSecretsFiles(config, consumerJava, keyUtility, keyProducer, bitHelper),
            javaProducersSecretsFiles
        );

        processProducers(
            finder.producerOpenCL,
            bitHelper::assertBatchSizeInBitsIsInRange,
            this::getKeyProducer,
            (config, keyProducer) -> new ProducerOpenCL(config, consumerJava, keyUtility, keyProducer, bitHelper),
            openCLProducers
        );
    }
    
    private <T extends CProducer, P> void processProducers(
        List<T> configs,
        java.util.function.Consumer<Integer> batchSizeAssert,
        Function<T, KeyProducer> getKeyProducer,
        BiFunction<T, KeyProducer, P> producerConstructor,
        Collection<P> targetCollection
    ) {
        if (configs != null) {
            for (T config : configs) {
                batchSizeAssert.accept(config.batchSizeInBits);
                KeyProducer keyProducer = getKeyProducer.apply(config);
                P producer = producerConstructor.apply(config, keyProducer);
                targetCollection.add(producer);
            }
        }
    }

    public KeyProducer getKeyProducer(CProducer cProducer) throws RuntimeException {
        KeyProducer keyProducer = keyProducers.get(cProducer.keyProducerId);
        if(keyProducer == null) {
            throw new KeyProducerIdUnknownException(cProducer.keyProducerId);
        }
        return keyProducer;
    }
    
    public void initProducer() {
        logger.info("initProducer");
        for (Producer producer : getAllProducers()) {
            producer.initProducer();
        }
    }
    
    public void startProducer() {
        logger.info("startProducer");
        for (Producer producer : getAllProducers()) {
            producerExecutorService.submit(producer);
        }
    }
    
    public void shutdownAndAwaitTermination() {
        logger.info("shutdownAndAwaitTermination");
        try {
            producerExecutorService.shutdown();
            producerExecutorService.awaitTermination(AWAIT_DURATION_TERMINATE.get(ChronoUnit.SECONDS), TimeUnit.SECONDS);
        } catch (InterruptedException ex) {
            throw new RuntimeException(ex);
        }
        
        // no producers are running anymore, the consumer can be interrupted
        if (consumerJava != null) {
            logger.info("Interrupt: " + consumerJava.toString());
            consumerJava.interrupt();
            consumerJava = null;
        }
        logger.info("consumerJava released.");
    }
    
    @Override
    public void interrupt() {
        logger.info("interrupt called: delegate interrupt to all keyProducers and producers");
        
        // Interrupt all Producers
        for (Producer producer : getAllProducers()) {
            logger.info("Interrupt Producer: " + producer.toString());
            producer.interrupt();
            logger.info("waitTillProducerNotRunning ...");
            producer.waitTillProducerNotRunning();
            producer.releaseProducer();
        }
        freeAllProducers();
        
        // Interrupt all KeyProducers
        for (KeyProducer keyProducer : getKeyProducers().values()) {
            logger.info("Interrupt KeyProducer: " + keyProducer.toString());
            keyProducer.interrupt();
        }
        freeAllKeyProducers();

        logger.info("All producers released and freed.");
    }
    
    public Map<String, KeyProducer> getKeyProducers() {
        return ImmutableMap.copyOf(keyProducers);
    }
    
    public List<Producer> getAllProducers() {
        List<Producer> producers = new ArrayList<>();
        producers.addAll(javaProducers);
        producers.addAll(javaProducersSecretsFiles);
        producers.addAll(openCLProducers);
        return producers;
    }
    
    public void freeAllProducers() {
        javaProducers.clear();
        javaProducersSecretsFiles.clear();
        openCLProducers.clear();
    }
    
    public void freeAllKeyProducers() {
        keyProducers.clear();
    }
    
    public List<Consumer> getAllConsumers() {
        List<Consumer> consumers = new ArrayList<>();
        if (consumerJava != null) {
            consumers.add(consumerJava);
        }
        return consumers;
    }
}

1	// @formatter:off
2	/**
3	* Copyright 2020 Bernard Ladenthin bernard.ladenthin@gmail.com
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License");
6	* you may not use this file except in compliance with the License.
7	* You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*
17	*/
18	// @formatter:on
19	package net.ladenthin.bitcoinaddressfinder;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import com.google.common.collect.ImmutableMap;
23	import java.time.Duration;
24	import java.time.temporal.ChronoUnit;
25	import java.util.ArrayList;
26	import java.util.Collection;
27	import java.util.HashMap;
28	import java.util.List;
29	import java.util.Map;
30	import java.util.concurrent.ExecutorService;
31	import java.util.concurrent.Executors;
32	import java.util.concurrent.TimeUnit;
33	import net.ladenthin.bitcoinaddressfinder.configuration.CFinder;
34	import net.ladenthin.bitcoinaddressfinder.configuration.CProducer;
35	import net.ladenthin.bitcoinaddressfinder.persistence.PersistenceUtils;
36	import org.bitcoinj.base.Network;
37	import org.jspecify.annotations.Nullable;
38	import org.slf4j.Logger;
39	import org.slf4j.LoggerFactory;
40	import java.util.function.*;
41
42	public class Finder implements Interruptable {
43
44	/**
45	* We must define a maximum time to wait for terminate. Wait for 100 thousand years is enough.
46	*/
47	@VisibleForTesting
48	static Duration AWAIT_DURATION_TERMINATE = Duration.ofDays(365L * 1000L);	1✔
49
50	protected Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
51
52	private final CFinder finder;
53
54	private final Map<String, KeyProducer> keyProducers = new HashMap<>();	1✔
55
56	@Nullable
57	private ConsumerJava consumerJava;
58
59	private final List<ProducerOpenCL> openCLProducers = new ArrayList<>();	1✔
60	private final List<ProducerJava> javaProducers = new ArrayList<>();	1✔
61	private final List<ProducerJavaSecretsFiles> javaProducersSecretsFiles = new ArrayList<>();	1✔
62
63	@VisibleForTesting	1✔
64	final ExecutorService producerExecutorService = Executors.newCachedThreadPool();	1✔
65
66	private final Network network = new NetworkParameterFactory().getNetwork();	1✔
67	private final KeyUtility keyUtility = new KeyUtility(network, new ByteBufferUtility(false));	1✔
68	private final PersistenceUtils persistenceUtils = new PersistenceUtils(network);	1✔
69	private final BitHelper bitHelper = new BitHelper();	1✔
70
71	public Finder(CFinder finder) {	1✔
72	this.finder = finder;	1✔
73	}	1✔
74
75	public void startKeyProducer() {
76	logger.info("startKeyProducer");	1✔
77	processKeyProducers(	1✔
78	finder.keyProducerJavaRandom,
79	cKeyProducerJavaRandom -> new KeyProducerJavaRandom(cKeyProducerJavaRandom, keyUtility, bitHelper),	1✔
80	cKeyProducerJavaRandom -> cKeyProducerJavaRandom.keyProducerId,	1✔
81	keyProducers
82	);
83
84	processKeyProducers(	1✔
85	finder.keyProducerJavaBip39,
86	cKeyProducerJavaBip39 -> new KeyProducerJavaBip39(cKeyProducerJavaBip39, keyUtility, bitHelper),	1✔
87	cKeyProducerJavaBip39 -> cKeyProducerJavaBip39.keyProducerId,	1✔
88	keyProducers
89	);
90
91	processKeyProducers(	1✔
92	finder.keyProducerJavaIncremental,
93	cKeyProducerJavaIncremental -> new KeyProducerJavaIncremental(cKeyProducerJavaIncremental, keyUtility, bitHelper),	1✔
94	cKeyProducerJavaIncremental -> cKeyProducerJavaIncremental.keyProducerId,	1✔
95	keyProducers
96	);
97
98	processKeyProducers(	1✔
99	finder.keyProducerJavaZmq,
100	cKeyProducerJavaZmq -> new KeyProducerJavaZmq(cKeyProducerJavaZmq, keyUtility, bitHelper),	1✔
101	cKeyProducerJavaZmq -> cKeyProducerJavaZmq.keyProducerId,	1✔
102	keyProducers
103	);
104	}	1✔
105
106	private <T, K> void processKeyProducers(
107	List<T> configList,
108	Function<T, K> constructor,
109	Function<T, String> getId,
110	Map<String, K> keyProducers
111	) {
112	if (configList != null) {	1✔
113	for (T config : configList) {	1✔
114	String keyProducerId = getId.apply(config);	1✔
115	if (keyProducerId == null) {	1✔
116	throw new KeyProducerIdNullException();	1✔
117	}
118	if (keyProducers.containsKey(keyProducerId)) {	1✔
119	throw new KeyProducerIdIsNotUniqueException(keyProducerId);	1✔
120	}
121	K keyProducer = constructor.apply(config);	1✔
122	keyProducers.put(keyProducerId, keyProducer);	1✔
123	}	1✔
124	}
125	}	1✔
126
127	public void startConsumer() {
128	logger.info("startConsumer");	1✔
129	if (finder.consumerJava != null) {	1✔
130	consumerJava = new ConsumerJava(finder.consumerJava, keyUtility, persistenceUtils);	1✔
131	consumerJava.initLMDB();	1✔
132	consumerJava.startConsumer();	1✔
133	consumerJava.startStatisticsTimer();	1✔
134	}
135	}	1✔
136
137	public void configureProducer() {
138	logger.info("configureProducer");	1✔
139	processProducers(	1✔
140	finder.producerJava,
141	bitHelper::assertBatchSizeInBitsIsInRange,	1✔
142	this::getKeyProducer,
143	(config, keyProducer) -> new ProducerJava(config, consumerJava, keyUtility, keyProducer, bitHelper),	1✔
144	javaProducers
145	);
146
147	processProducers(	1✔
148	finder.producerJavaSecretsFiles,
149	bitHelper::assertBatchSizeInBitsIsInRange,	1✔
150	this::getKeyProducer,
151	(config, keyProducer) -> new ProducerJavaSecretsFiles(config, consumerJava, keyUtility, keyProducer, bitHelper),	1✔
152	javaProducersSecretsFiles
153	);
154
155	processProducers(	1✔
156	finder.producerOpenCL,
157	bitHelper::assertBatchSizeInBitsIsInRange,	1✔
158	this::getKeyProducer,
159	(config, keyProducer) -> new ProducerOpenCL(config, consumerJava, keyUtility, keyProducer, bitHelper),	1✔
160	openCLProducers
161	);
162	}	1✔
163
164	private <T extends CProducer, P> void processProducers(
165	List<T> configs,
166	java.util.function.Consumer<Integer> batchSizeAssert,
167	Function<T, KeyProducer> getKeyProducer,
168	BiFunction<T, KeyProducer, P> producerConstructor,
169	Collection<P> targetCollection
170	) {
171	if (configs != null) {	1✔
172	for (T config : configs) {	1✔
173	batchSizeAssert.accept(config.batchSizeInBits);	1✔
174	KeyProducer keyProducer = getKeyProducer.apply(config);	1✔
175	P producer = producerConstructor.apply(config, keyProducer);	1✔
176	targetCollection.add(producer);	1✔
177	}	1✔
178	}
179	}	1✔
180
181	public KeyProducer getKeyProducer(CProducer cProducer) throws RuntimeException {
182	KeyProducer keyProducer = keyProducers.get(cProducer.keyProducerId);	1✔
183	if(keyProducer == null) {	1✔
184	throw new KeyProducerIdUnknownException(cProducer.keyProducerId);	1✔
185	}
186	return keyProducer;	1✔
187	}
188
189	public void initProducer() {
190	logger.info("initProducer");	1✔
191	for (Producer producer : getAllProducers()) {	1✔
192	producer.initProducer();	1✔
193	}	1✔
194	}	1✔
195
196	public void startProducer() {
197	logger.info("startProducer");	1✔
198	for (Producer producer : getAllProducers()) {	1✔
199	producerExecutorService.submit(producer);	1✔
200	}	1✔
201	}	1✔
202
203	public void shutdownAndAwaitTermination() {
204	logger.info("shutdownAndAwaitTermination");	1✔
205	try {
206	producerExecutorService.shutdown();	1✔
207	producerExecutorService.awaitTermination(AWAIT_DURATION_TERMINATE.get(ChronoUnit.SECONDS), TimeUnit.SECONDS);	1✔
UNCOV 208	} catch (InterruptedException ex) {	×
UNCOV 209	throw new RuntimeException(ex);	×
210	}	1✔
211
212	// no producers are running anymore, the consumer can be interrupted
213	if (consumerJava != null) {	1✔
214	logger.info("Interrupt: " + consumerJava.toString());	1✔
215	consumerJava.interrupt();	1✔
216	consumerJava = null;	1✔
217	}
218	logger.info("consumerJava released.");	1✔
219	}	1✔
220
221	@Override
222	public void interrupt() {
223	logger.info("interrupt called: delegate interrupt to all keyProducers and producers");	1✔
224
225	// Interrupt all Producers
226	for (Producer producer : getAllProducers()) {	1✔
227	logger.info("Interrupt Producer: " + producer.toString());	1✔
228	producer.interrupt();	1✔
229	logger.info("waitTillProducerNotRunning ...");	1✔
230	producer.waitTillProducerNotRunning();	1✔
231	producer.releaseProducer();	1✔
232	}	1✔
233	freeAllProducers();	1✔
234
235	// Interrupt all KeyProducers
236	for (KeyProducer keyProducer : getKeyProducers().values()) {	1✔
237	logger.info("Interrupt KeyProducer: " + keyProducer.toString());	1✔
238	keyProducer.interrupt();	1✔
239	}	1✔
240	freeAllKeyProducers();	1✔
241
242	logger.info("All producers released and freed.");	1✔
243	}	1✔
244
245	public Map<String, KeyProducer> getKeyProducers() {
246	return ImmutableMap.copyOf(keyProducers);	1✔
247	}
248
249	public List<Producer> getAllProducers() {
250	List<Producer> producers = new ArrayList<>();	1✔
251	producers.addAll(javaProducers);	1✔
252	producers.addAll(javaProducersSecretsFiles);	1✔
253	producers.addAll(openCLProducers);	1✔
254	return producers;	1✔
255	}
256
257	public void freeAllProducers() {
258	javaProducers.clear();	1✔
259	javaProducersSecretsFiles.clear();	1✔
260	openCLProducers.clear();	1✔
261	}	1✔
262
263	public void freeAllKeyProducers() {
264	keyProducers.clear();	1✔
265	}	1✔
266
267	public List<Consumer> getAllConsumers() {
268	List<Consumer> consumers = new ArrayList<>();	1✔
269	if (consumerJava != null) {	1✔
270	consumers.add(consumerJava);	1✔
271	}
272	return consumers;	1✔
273	}
274	}

bernardladenthin / BitcoinAddressFinder / #353

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous