Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

added methods for support ValueProvider in MqttIO #32181

Closed
wants to merge 1 commit into from
Closed

added methods for support ValueProvider in MqttIO #32181

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 67 additions & 16 deletions sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.UnboundedSource;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
Expand Down Expand Up @@ -123,29 +124,29 @@ private MqttIO() {}
@AutoValue
public abstract static class ConnectionConfiguration implements Serializable {

abstract String getServerUri();
abstract ValueProvider<String> getServerUri();

abstract String getTopic();
abstract ValueProvider<String> getTopic();

abstract @Nullable String getClientId();
abstract @Nullable ValueProvider<String> getClientId();

abstract @Nullable String getUsername();
abstract @Nullable ValueProvider<String> getUsername();

abstract @Nullable String getPassword();
abstract @Nullable ValueProvider<String> getPassword();

abstract Builder builder();

@AutoValue.Builder
abstract static class Builder {
abstract Builder setServerUri(String serverUri);
abstract Builder setServerUri(ValueProvider<String> serverUri);

abstract Builder setTopic(String topic);
abstract Builder setTopic(ValueProvider<String> topic);

abstract Builder setClientId(String clientId);
abstract Builder setClientId(ValueProvider<String> clientId);

abstract Builder setUsername(String username);
abstract Builder setUsername(ValueProvider<String> username);

abstract Builder setPassword(String password);
abstract Builder setPassword(ValueProvider<String> password);

abstract ConnectionConfiguration build();
}
Expand All @@ -161,6 +162,23 @@ abstract static class Builder {
public static ConnectionConfiguration create(String serverUri, String topic) {
checkArgument(serverUri != null, "serverUri can not be null");
checkArgument(topic != null, "topic can not be null");
return create(
ValueProvider.StaticValueProvider.of(serverUri),
ValueProvider.StaticValueProvider.of(topic));
}

/**
* Describe a connection configuration to the MQTT broker. This method creates a unique random
* MQTT client ID.
*
* @param serverUri The MQTT broker URI.
* @param topic The MQTT getTopic pattern.
* @return A connection configuration to the MQTT broker.
*/
public static ConnectionConfiguration create(
ValueProvider<String> serverUri, ValueProvider<String> topic) {
checkArgument(serverUri != null, "serverUri can not be null");
checkArgument(topic != null, "topic can not be null");
return new AutoValue_MqttIO_ConnectionConfiguration.Builder()
.setServerUri(serverUri)
.setTopic(topic)
Expand All @@ -169,28 +187,56 @@ public static ConnectionConfiguration create(String serverUri, String topic) {

/** Set up the MQTT broker URI. */
public ConnectionConfiguration withServerUri(String serverUri) {
checkArgument(serverUri != null, "serverUri can not be null");
return withServerUri(ValueProvider.StaticValueProvider.of(serverUri));
}

/** Set up the MQTT broker URI. */
public ConnectionConfiguration withServerUri(ValueProvider<String> serverUri) {
checkArgument(serverUri != null, "serverUri can not be null");
return builder().setServerUri(serverUri).build();
}

/** Set up the MQTT getTopic pattern. */
public ConnectionConfiguration withTopic(String topic) {
checkArgument(topic != null, "topic can not be null");
return withTopic(ValueProvider.StaticValueProvider.of(topic));
}

/** Set up the MQTT getTopic pattern. */
public ConnectionConfiguration withTopic(ValueProvider<String> topic) {
checkArgument(topic != null, "topic can not be null");
return builder().setTopic(topic).build();
}

/** Set up the client ID prefix, which is used to construct a unique client ID. */
public ConnectionConfiguration withClientId(String clientId) {
checkArgument(clientId != null, "clientId can not be null");
return withClientId(ValueProvider.StaticValueProvider.of(clientId));
}

/** Set up the client ID prefix, which is used to construct a unique client ID. */
public ConnectionConfiguration withClientId(ValueProvider<String> clientId) {
checkArgument(clientId != null, "clientId can not be null");
return builder().setClientId(clientId).build();
}

public ConnectionConfiguration withUsername(String username) {
checkArgument(username != null, "username can not be null");
return withUsername(ValueProvider.StaticValueProvider.of(username));
}

public ConnectionConfiguration withUsername(ValueProvider<String> username) {
checkArgument(username != null, "username can not be null");
return builder().setUsername(username).build();
}

public ConnectionConfiguration withPassword(String password) {
checkArgument(password != null, "password can not be null");
return withPassword(ValueProvider.StaticValueProvider.of(password));
}

public ConnectionConfiguration withPassword(ValueProvider<String> password) {
checkArgument(password != null, "password can not be null");
return builder().setPassword(password).build();
}
Expand All @@ -205,12 +251,15 @@ private void populateDisplayData(DisplayData.Builder builder) {
private MQTT createClient() throws Exception {
LOG.debug("Creating MQTT client to {}", getServerUri());
MQTT client = new MQTT();
client.setHost(getServerUri());
client.setHost(getServerUri().get());
if (getUsername() != null) {
LOG.debug("MQTT client uses username {}", getUsername());
client.setUserName(getUsername());
client.setPassword(getPassword());
LOG.debug("MQTT client uses username {}", getUsername().get());
client.setUserName(getUsername().get());
}
if (getPassword() != null) {
client.setPassword(getPassword().get());
}

if (getClientId() != null) {
String clientId = getClientId() + "-" + UUID.randomUUID().toString();
clientId =
Expand Down Expand Up @@ -434,7 +483,9 @@ public boolean start() throws IOException {
connection = client.blockingConnection();
connection.connect();
connection.subscribe(
new Topic[] {new Topic(spec.connectionConfiguration().getTopic(), QoS.AT_LEAST_ONCE)});
new Topic[] {
new Topic(spec.connectionConfiguration().getTopic().get(), QoS.AT_LEAST_ONCE)
});
return advance();
} catch (Exception e) {
throw new IOException(e);
Expand Down Expand Up @@ -578,7 +629,7 @@ public void processElement(ProcessContext context) throws Exception {
byte[] payload = context.element();
LOG.debug("Sending message {}", new String(payload, StandardCharsets.UTF_8));
connection.publish(
spec.connectionConfiguration().getTopic(), payload, QoS.AT_LEAST_ONCE, false);
spec.connectionConfiguration().getTopic().get(), payload, QoS.AT_LEAST_ONCE, false);
}

@Teardown
Expand Down
Loading