org.apache.beam.sdk.io.solace.broker

Class SessionService

java.lang.Object

org.apache.beam.sdk.io.solace.broker.SessionService

All Implemented Interfaces:

java.io.Serializable

Direct Known Subclasses:

BasicAuthJcsmpSessionService

public abstract class SessionService
extends java.lang.Object
implements java.io.Serializable

The SessionService interface provides a set of methods for managing a session with the Solace messaging system. It allows for establishing a connection, creating a message-receiver object, checking if the connection is closed or not, and gracefully closing the session.

Override this class and the method initializeSessionProperties(JCSMPProperties) with your specific properties, including all those related to authentication.

The connector will call the method only once per session created, so you can perform relatively heavy operations in that method (e.g. connect to a store or vault to retrieve credentials).

There are some default properties that are set by default and can be overridden in this provider, that are relevant for the writer connector, and not used in the case of the read connector (since they are not necessary for reading):

• VPN_NAME: default
• GENERATE_SEND_TIMESTAMPS: true
• PUB_MULTI_THREAD: true

The connector overrides other properties, regardless of what this provider sends to the connector. Those properties are the following. Again, these properties are only relevant for the write connector.

• PUB_ACK_WINDOW_SIZE
• MESSAGE_CALLBACK_ON_REACTOR

Those properties are set by the connector based on the values of SolaceIO.Write.withWriterType(SolaceIO.WriterType) and SolaceIO.Write.withSubmissionMode(SolaceIO.SubmissionMode).

The method will always run in a worker thread or task, and not in the driver program. If you need to access any resource to set the properties, you need to make sure that the worker has the network connectivity required for that, and that any credential or configuration is passed to the provider through the constructor.

The connector ensures that no two threads will be calling that method at the same time, so you don't have to take any specific precautions to avoid race conditions.

For basic authentication, use BasicAuthJcsmpSessionService and BasicAuthJcsmpSessionServiceFactory.

For other situations, you need to extend this class and implement the `equals` method, so two instances of your class can be compared by value. We recommend using AutoValue for that. For instance:

 {@literal }@AutoValue
 public class MySessionService extends SessionService {
   abstract String authToken();

   public static MySessionService create(String authToken) {
       return new AutoValue_MySessionService(authToken);
   }

   {@literal }@Override
   public JCSMPProperties initializeSessionProperties(JCSMPProperties baseProps) {
     baseProps.setProperty(JCSMPProperties.AUTHENTICATION_SCHEME, JCSMPProperties.AUTHENTICATION_SCHEME_OAUTH2);
     baseProps.setProperty(JCSMPProperties.OAUTH2_ACCESS_TOKEN, authToken());
     return props;
   }

   {@literal }@Override
   public void connect() {
       ...
   }

   ...
 }
 

See Also:

Serialized Form

Field Summary

Fields

Modifier and Type	Field and Description
static java.lang.String	DEFAULT_VPN_NAME

Constructor Summary

Constructors

Constructor and Description
SessionService()

Method Summary

Modifier and Type	Method and Description
abstract void	close() Gracefully closes the connection to the service.
abstract void	connect() Establishes a connection to the service.
abstract boolean	equals(@Nullable java.lang.Object other) You need to override this method to be able to compare these objects by value.
abstract MessageProducer	getInitializedProducer(SolaceIO.SubmissionMode mode) Returns a MessageProducer object for publishing messages to Solace.
abstract java.util.Queue<Solace.PublishResult>	getPublishedResultsQueue() Returns the Queue instance associated with this session, with the asynchronously received callbacks from Solace for message publications.
abstract MessageReceiver	getReceiver() Returns a MessageReceiver object for receiving messages from Solace.
abstract int	hashCode() You need to override this method to be able to compare these objects by value.
abstract JCSMPProperties	initializeSessionProperties(JCSMPProperties baseProperties) Override this method and provide your specific properties, including all those related to authentication, and possibly others too.
JCSMPProperties	initializeWriteSessionProperties(SolaceIO.SubmissionMode mode) This method will be called by the write connector when a new session is started.
abstract boolean	isClosed() Checks whether the connection to the service is currently closed.

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, toString, wait, wait, wait

Field Detail

DEFAULT_VPN_NAME

public static final java.lang.String DEFAULT_VPN_NAME

See Also:

Constant Field Values

Constructor Detail

SessionService

public SessionService()

Method Detail

connect

public abstract void connect()

Establishes a connection to the service. This could involve providing connection details like host, port, VPN name, username, and password.

close

public abstract void close()

Gracefully closes the connection to the service.

isClosed

public abstract boolean isClosed()

Checks whether the connection to the service is currently closed. This method is called when an `UnboundedSolaceReader` is starting to read messages - a session will be created if this returns true.

getReceiver

public abstract MessageReceiver getReceiver()

Returns a MessageReceiver object for receiving messages from Solace. If it is the first time this method is used, the receiver is created from the session instance, otherwise it returns the receiver created initially.

getInitializedProducer

public abstract MessageProducer getInitializedProducer(SolaceIO.SubmissionMode mode)

Returns a MessageProducer object for publishing messages to Solace. If it is the first time this method is used, the producer is created from the session instance, otherwise it returns the producer created initially.

getPublishedResultsQueue

public abstract java.util.Queue<Solace.PublishResult> getPublishedResultsQueue()

Returns the Queue instance associated with this session, with the asynchronously received callbacks from Solace for message publications. The queue implementation has to be thread-safe for production use-cases.

initializeSessionProperties

public abstract JCSMPProperties initializeSessionProperties(JCSMPProperties baseProperties)

Override this method and provide your specific properties, including all those related to authentication, and possibly others too. The baseProperties parameter sets the Solace VPN to "default" if none is specified.

You should add your properties to the parameter baseProperties, and return the result.

The method will be used whenever the session needs to be created or refreshed. If you are setting credentials with expiration, just make sure that the latest available credentials (e.g. renewed token) are set when the method is called.

For a list of all the properties that can be set, please check the following link:

• https://docs.solace.com/API-Developer-Online-Ref-Documentation/java/com/solacesystems/jcsmp/JCSMPProperties.html

equals

public abstract boolean equals(@Nullable java.lang.Object other)

You need to override this method to be able to compare these objects by value. We recommend using AutoValue for that.

Overrides:

equals in class java.lang.Object

hashCode

public abstract int hashCode()

You need to override this method to be able to compare these objects by value. We recommend using AutoValue for that.

Overrides:

hashCode in class java.lang.Object

initializeWriteSessionProperties

public final JCSMPProperties initializeWriteSessionProperties(SolaceIO.SubmissionMode mode)

This method will be called by the write connector when a new session is started.

This call will happen in the worker, so you need to make sure that the worker has access to the resources you need to set the properties.

The call will happen only once per session initialization. Typically, that will be when the worker and the client are created. But if for any reason the session is lost (e.g. expired auth token), this method will be called again.