BAEL-1486 - sample code for JGroups. Fixed a typo in InfluxDB. (#3578)
This commit is contained in:
Eric Goebelbecker
maibin
parent
8b7630335a
commit
bf5b3045a1
|
|
@ -2,7 +2,6 @@
|
|||
|
||||
### Relevant Article:
|
||||
- [Introduction to using InfluxDB with Java](http://www.baeldung.com/using-influxdb-with-java/)
|
||||
- [Using InfluxDB with Java](http://www.baeldung.com/java-influxdb)
|
||||
|
||||
### Overview
|
||||
This Maven project contains the Java code for the article linked above.
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import org.influxdb.dto.*;
|
|||
import org.influxdb.impl.InfluxDBResultMapper;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
|
|
@ -103,7 +102,7 @@ public class InfluxDBConnectionLiveTest {
|
|||
// another brief pause.
|
||||
Thread.sleep(10);
|
||||
|
||||
List<MemoryPoint> memoryPointList = getPoints(connection, "Select * from memory", "baeldung");
|
||||
List<com.baeldung.influxdb.MemoryPoint> memoryPointList = getPoints(connection, "Select * from memory", "baeldung");
|
||||
|
||||
assertEquals(10, memoryPointList.size());
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
## Reliable Messaging with JGroups Tutorial Project
|
||||
|
||||
### Relevant Article:
|
||||
- [Reliable Messaging with JGroups](http://www.baeldung.com/reliable-messaging-with-jgroups/)
|
||||
|
||||
### Overview
|
||||
This Maven project contains the Java code for the article linked above.
|
||||
|
||||
### Package Organization
|
||||
Java classes for the intro tutorial are in the org.baeldung.jgroups package.
|
||||
|
||||
|
||||
### Running the tests
|
||||
|
||||
```
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>jgroups</artifactId>
|
||||
<version>0.1-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
<name>jgroups</name>
|
||||
<description>Reliable Messaging with JGroups Tutorial</description>
|
||||
|
||||
<parent>
|
||||
<groupId>com.baeldung</groupId>
|
||||
<artifactId>parent-modules</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.jgroups</groupId>
|
||||
<artifactId>jgroups</artifactId>
|
||||
<version>4.0.10.Final</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>commons-cli</groupId>
|
||||
<artifactId>commons-cli</artifactId>
|
||||
<version>1.4</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<properties>
|
||||
<java.version>1.8</java.version>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,222 @@
|
|||
package com.baeldung.jgroups;
|
||||
|
||||
import org.apache.commons.cli.*;
|
||||
import org.jgroups.*;
|
||||
import org.jgroups.util.Util;
|
||||
|
||||
import java.io.*;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public class JGroupsMessenger extends ReceiverAdapter {
|
||||
|
||||
private JChannel channel;
|
||||
private String userName;
|
||||
private String clusterName;
|
||||
private View lastView;
|
||||
private boolean running = true;
|
||||
|
||||
// Our shared state
|
||||
private Integer messageCount = 0;
|
||||
|
||||
/**
|
||||
* Connect to a JGroups cluster using command line options
|
||||
* @param args command line arguments
|
||||
* @throws Exception
|
||||
*/
|
||||
private void start(String[] args) throws Exception {
|
||||
processCommandline(args);
|
||||
|
||||
// Create the channel
|
||||
// This file could be moved, or made a command line option.
|
||||
channel = new JChannel("src/main/resources/udp.xml");
|
||||
|
||||
// Set a name
|
||||
channel.name(userName);
|
||||
|
||||
// Register for callbacks
|
||||
channel.setReceiver(this);
|
||||
|
||||
// Ignore our messages
|
||||
channel.setDiscardOwnMessages(true);
|
||||
|
||||
// Connect
|
||||
channel.connect(clusterName);
|
||||
|
||||
// Start state transfer
|
||||
channel.getState(null, 0);
|
||||
|
||||
// Do the things
|
||||
processInput();
|
||||
|
||||
// Clean up
|
||||
channel.close();
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Quick and dirty implementaton of commons cli for command line args
|
||||
* @param args the command line args
|
||||
* @throws ParseException
|
||||
*/
|
||||
private void processCommandline(String[] args) throws ParseException {
|
||||
|
||||
// Options, parser, friendly help
|
||||
Options options = new Options();
|
||||
CommandLineParser parser = new DefaultParser();
|
||||
HelpFormatter formatter = new HelpFormatter();
|
||||
|
||||
options.addOption("u", "user", true, "User name")
|
||||
.addOption("c", "cluster", true, "Cluster name");
|
||||
|
||||
CommandLine line = parser.parse(options, args);
|
||||
|
||||
if (line.hasOption("user")) {
|
||||
userName = line.getOptionValue("user");
|
||||
} else {
|
||||
formatter.printHelp("JGroupsMessenger: need a user name.\n", options);
|
||||
System.exit(-1);
|
||||
}
|
||||
|
||||
if (line.hasOption("cluster")) {
|
||||
clusterName = line.getOptionValue("cluster");
|
||||
} else {
|
||||
formatter.printHelp("JGroupsMessenger: need a cluster name.\n", options);
|
||||
System.exit(-1);
|
||||
}
|
||||
}
|
||||
|
||||
// Start it up
|
||||
public static void main(String[] args) throws Exception {
|
||||
new JGroupsMessenger().start(args);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void viewAccepted(View newView) {
|
||||
|
||||
// Save view if this is the first
|
||||
if (lastView == null) {
|
||||
System.out.println("Received initial view:");
|
||||
newView.forEach(System.out::println);
|
||||
} else {
|
||||
|
||||
// Compare to last view
|
||||
System.out.println("Received new view.");
|
||||
|
||||
List<Address> newMembers = View.newMembers(lastView, newView);
|
||||
if (newMembers.size() > 0) {
|
||||
System.out.println("New members: ");
|
||||
newMembers.forEach(System.out::println);
|
||||
}
|
||||
|
||||
List<Address> exMembers = View.leftMembers(lastView, newView);
|
||||
if (exMembers.size() > 0) {
|
||||
System.out.println("Exited members:");
|
||||
exMembers.forEach(System.out::println);
|
||||
}
|
||||
}
|
||||
lastView = newView;
|
||||
}
|
||||
|
||||
/**
|
||||
* Loop on console input until we see 'x' to exit
|
||||
*/
|
||||
private void processInput() {
|
||||
|
||||
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
|
||||
while (running) {
|
||||
try {
|
||||
|
||||
// Get a destination, <enter> means broadcast
|
||||
Address destination = null;
|
||||
System.out.print("Enter a destination: ");
|
||||
System.out.flush();
|
||||
String destinationName = in.readLine().toLowerCase();
|
||||
|
||||
if (destinationName.equals("x")) {
|
||||
running = false;
|
||||
continue;
|
||||
} else if (!destinationName.isEmpty()) {
|
||||
Optional<Address> optDestination = getAddress(destinationName);
|
||||
if (optDestination.isPresent()) {
|
||||
destination = optDestination.get();
|
||||
} else {
|
||||
System.out.println("Destination not found, try again.");
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// Accept a string to send
|
||||
System.out.print("Enter a message: ");
|
||||
System.out.flush();
|
||||
String line = in.readLine().toLowerCase();
|
||||
sendMessage(destination, line);
|
||||
} catch (IOException ioe) {
|
||||
running = false;
|
||||
}
|
||||
}
|
||||
System.out.println("Exiting.");
|
||||
}
|
||||
|
||||
/**
|
||||
* Send message from here
|
||||
* @param destination the destination
|
||||
* @param messageString the message
|
||||
*/
|
||||
private void sendMessage(Address destination, String messageString) {
|
||||
try {
|
||||
System.out.println("Sending " + messageString + " to " + destination);
|
||||
Message message = new Message(destination, messageString);
|
||||
channel.send(message);
|
||||
} catch (Exception exception) {
|
||||
System.err.println("Exception sending message: " + exception.getMessage());
|
||||
running = false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void receive(Message message) {
|
||||
// Print source and dest with message
|
||||
String line = "Message received from: " + message.getSrc() + " to: " + message.getDest() + " -> " + message.getObject();
|
||||
|
||||
// Only track the count of broadcast messages
|
||||
// Tracking direct message would make for a pointless state
|
||||
if (message.getDest() == null) {
|
||||
messageCount++;
|
||||
System.out.println("Message count: " + messageCount);
|
||||
}
|
||||
|
||||
System.out.println(line);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void getState(OutputStream output) throws Exception {
|
||||
// Serialize into the stream
|
||||
Util.objectToStream(messageCount, new DataOutputStream(output));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setState(InputStream input) {
|
||||
|
||||
// NOTE: since we know that incrementing the count and transferring the state
|
||||
// is done inside the JChannel's thread, we don't have to worry about synchronizing
|
||||
// messageCount. For production code it should be synchronized!
|
||||
try {
|
||||
// Deserialize
|
||||
messageCount = Util.objectFromStream(new DataInputStream(input));
|
||||
} catch (Exception e) {
|
||||
System.out.println("Error deserialing state!");
|
||||
}
|
||||
System.out.println(messageCount + " is the current messagecount.");
|
||||
}
|
||||
|
||||
|
||||
private Optional<Address> getAddress(String name) {
|
||||
View view = channel.view();
|
||||
return view.getMembers().stream().filter(address -> name.equals(address.toString())).findFirst();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
<config xmlns="urn:org:jgroups"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
|
||||
<UDP
|
||||
mcast_port="${jgroups.udp.mcast_port:45588}"
|
||||
ip_ttl="4"
|
||||
tos="8"
|
||||
ucast_recv_buf_size="5M"
|
||||
ucast_send_buf_size="5M"
|
||||
mcast_recv_buf_size="5M"
|
||||
mcast_send_buf_size="5M"
|
||||
max_bundle_size="64K"
|
||||
enable_diagnostics="true"
|
||||
thread_naming_pattern="cl"
|
||||
|
||||
thread_pool.min_threads="0"
|
||||
thread_pool.max_threads="20"
|
||||
thread_pool.keep_alive_time="30000"/>
|
||||
|
||||
<PING />
|
||||
<MERGE3 max_interval="30000"
|
||||
min_interval="10000"/>
|
||||
<FD_SOCK/>
|
||||
<FD_ALL/>
|
||||
<VERIFY_SUSPECT timeout="1500" />
|
||||
<BARRIER />
|
||||
<pbcast.NAKACK2 xmit_interval="500"
|
||||
xmit_table_num_rows="100"
|
||||
xmit_table_msgs_per_row="2000"
|
||||
xmit_table_max_compaction_time="30000"
|
||||
use_mcast_xmit="false"
|
||||
discard_delivered_msgs="true"/>
|
||||
<UNICAST3 xmit_interval="500"
|
||||
xmit_table_num_rows="100"
|
||||
xmit_table_msgs_per_row="2000"
|
||||
xmit_table_max_compaction_time="60000"
|
||||
conn_expiry_timeout="0"/>
|
||||
<pbcast.STABLE desired_avg_gossip="50000"
|
||||
max_bytes="4M"/>
|
||||
<pbcast.GMS print_local_addr="true" join_timeout="2000"/>
|
||||
<UFC max_credits="2M"
|
||||
min_threshold="0.4"/>
|
||||
<MFC max_credits="2M"
|
||||
min_threshold="0.4"/>
|
||||
<FRAG2 frag_size="60K" />
|
||||
<RSVP resend_interval="2000" timeout="10000"/>
|
||||
<pbcast.STATE_TRANSFER />
|
||||
</config>
|
||||
Loading…
Reference in New Issue