#358

Committed 25 Jul 2025 10:40PM UTC coverage: 72.611% (+0.2%) from 72.394%

Build # #358

Build Type

push

Committed by

Commit Message

Extract AbstractKeyProducerQueueBuffered

Run Details

101 of 108 new or added lines in 8 files covered. (93.52%)

2 existing lines in 1 file now uncovered.

1649 of 2271 relevant lines covered (72.61%)

0.73 hits per line

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

93.94

/src/main/java/net/ladenthin/bitcoinaddressfinder/keyproducer/KeyProducerJavaZmq.java

// @formatter:off
/**
 * Copyright 2025 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.keyproducer;

import net.ladenthin.bitcoinaddressfinder.BitHelper;
import net.ladenthin.bitcoinaddressfinder.KeyUtility;
import net.ladenthin.bitcoinaddressfinder.configuration.CKeyProducerJavaZmq;
import net.ladenthin.bitcoinaddressfinder.PublicKeyBytes;
import org.slf4j.Logger;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;

public class KeyProducerJavaZmq extends AbstractKeyProducerQueueBuffered<CKeyProducerJavaZmq> {

    private final ZContext context;
    private final ZMQ.Socket socket;
    private final Thread receiverThread;
    
    public KeyProducerJavaZmq(CKeyProducerJavaZmq config, KeyUtility keyUtility, BitHelper bitHelper, Logger logger) {
        super(config, keyUtility, logger);

        context = new ZContext();
        socket = context.createSocket(SocketType.PULL);

        if (cKeyProducerJava.mode == CKeyProducerJavaZmq.Mode.BIND) {
            socket.bind(cKeyProducerJava.address);
        } else {
            socket.connect(cKeyProducerJava.address);
        }

        // Fallback to 1000 ms if timeout not set
        int internalTimeout = cKeyProducerJava.timeout > 0 ? cKeyProducerJava.timeout : 1000;
        socket.setReceiveTimeOut(internalTimeout);

        receiverThread = new Thread(() -> {
            while (!shouldStop && !Thread.currentThread().isInterrupted()) {
                try {
                    byte[] msg = socket.recv(0); // blocking up to timeout
                    if (msg != null) {
                        if (msg.length == PublicKeyBytes.PRIVATE_KEY_MAX_NUM_BYTES) {
                            addSecret(msg);
                        } else {
                            System.err.println("Received invalid secret length: " + msg.length);
                        }
                    }
                    // if msg is null: it's a timeout — just loop again
                } catch (ZMQException e) {
                    if (shouldStop || e.getErrorCode() == ZMQ.Error.ETERM.getCode()) break;
                    e.printStackTrace(); // unexpected ZMQ errors
                }
            }
        }, "ZMQ-Receiver");

        receiverThread.setDaemon(true);
        receiverThread.start();
    }

    @Override
    protected int getReadTimeout() {
        return cKeyProducerJava.timeout > 0 ? cKeyProducerJava.timeout : 1000;
    }

    @Override
    public void interrupt() {
        shouldStop = true;
        receiverThread.interrupt(); // allow thread to exit if blocked
    }

    public void close() {
        interrupt();
        try {
            receiverThread.join(500);
        } catch (InterruptedException ignored) {}
        socket.close();
        context.close();
    }
}

1	// @formatter:off
2	/**
3	* Copyright 2025 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.keyproducer;
20
21	import net.ladenthin.bitcoinaddressfinder.BitHelper;
22	import net.ladenthin.bitcoinaddressfinder.KeyUtility;
23	import net.ladenthin.bitcoinaddressfinder.configuration.CKeyProducerJavaZmq;
24	import net.ladenthin.bitcoinaddressfinder.PublicKeyBytes;
25	import org.slf4j.Logger;
26	import org.zeromq.SocketType;
27	import org.zeromq.ZContext;
28	import org.zeromq.ZMQ;
29	import org.zeromq.ZMQException;
30
31	public class KeyProducerJavaZmq extends AbstractKeyProducerQueueBuffered<CKeyProducerJavaZmq> {
32
33	private final ZContext context;
34	private final ZMQ.Socket socket;
35	private final Thread receiverThread;
36
37	public KeyProducerJavaZmq(CKeyProducerJavaZmq config, KeyUtility keyUtility, BitHelper bitHelper, Logger logger) {
38	super(config, keyUtility, logger);	1✔
39
40	context = new ZContext();	1✔
41	socket = context.createSocket(SocketType.PULL);	1✔
42
43	if (cKeyProducerJava.mode == CKeyProducerJavaZmq.Mode.BIND) {	1✔
44	socket.bind(cKeyProducerJava.address);	1✔
45	} else {
46	socket.connect(cKeyProducerJava.address);	1✔
47	}
48
49	// Fallback to 1000 ms if timeout not set
50	int internalTimeout = cKeyProducerJava.timeout > 0 ? cKeyProducerJava.timeout : 1000;	1✔
51	socket.setReceiveTimeOut(internalTimeout);	1✔
52
53	receiverThread = new Thread(() -> {	1✔
54	while (!shouldStop && !Thread.currentThread().isInterrupted()) {	1✔
55	try {
56	byte[] msg = socket.recv(0); // blocking up to timeout	1✔
57	if (msg != null) {	1✔
58	if (msg.length == PublicKeyBytes.PRIVATE_KEY_MAX_NUM_BYTES) {	1✔
59	addSecret(msg);	1✔
60	} else {
NEW 61	System.err.println("Received invalid secret length: " + msg.length);	×
62	}
63	}
64	// if msg is null: it's a timeout — just loop again
65	} catch (ZMQException e) {	1✔
66	if (shouldStop \|\| e.getErrorCode() == ZMQ.Error.ETERM.getCode()) break;	1✔
NEW 67	e.printStackTrace(); // unexpected ZMQ errors	×
68	}	1✔
69	}
70	}, "ZMQ-Receiver");	1✔
71
72	receiverThread.setDaemon(true);	1✔
73	receiverThread.start();	1✔
74	}	1✔
75
76	@Override
77	protected int getReadTimeout() {
78	return cKeyProducerJava.timeout > 0 ? cKeyProducerJava.timeout : 1000;	1✔
79	}
80
81	@Override
82	public void interrupt() {
83	shouldStop = true;	1✔
84	receiverThread.interrupt(); // allow thread to exit if blocked	1✔
85	}	1✔
86
87	public void close() {
88	interrupt();	1✔
89	try {
90	receiverThread.join(500);	1✔
91	} catch (InterruptedException ignored) {}	1✔
92	socket.close();	1✔
93	context.close();	1✔
94	}	1✔
95	}

bernardladenthin / BitcoinAddressFinder / #358

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