public abstract class AbstractMessageReceiver extends BasicComponent implements StatisticsContainer, MessageReceiver
MUC
,
PubSub
, SessionManager
.
The class offers scripting API for administrator ad-hoc commands.
By default it internally uses priority queues which in some rare cases may
lead to packets reordering. When this happens and it is unacceptable for the
deployment non-priority queues can be used. The queues size is limited and
depends on the available memory size.
Packets are processed by processPacket(Packet packet)
method
which is concurrently called from multiple threads.
Created: Tue Nov 22 07:07:11 2005Modifier and Type | Field and Description |
---|---|
protected static long |
HOUR
Constant used in time calculation procedures.
|
static String |
INCOMING_FILTERS_PROP_KEY
Configuration property key for setting incoming packets filters on the
component level.
|
static String |
INCOMING_FILTERS_PROP_VAL
Configuration property default vakue with a default incoming packet filter
loaded by Tigase server.
|
static String |
MAX_QUEUE_SIZE_PROP_KEY
Configuration property key allowing to overwrite a default (memory size
dependent) size for the component internal queues.
|
static Integer |
MAX_QUEUE_SIZE_PROP_VAL
A default value for max queue size property.
|
protected int |
maxInQueueSize |
protected int |
maxOutQueueSize |
protected int |
maxQueueSize
Field description
|
protected static long |
MINUTE
Constant used in time calculation procedures.
|
static String |
OUTGOING_FILTERS_PROP_KEY
Configuration property key for setting outgoing packets filters on the
component level.
|
static String |
OUTGOING_FILTERS_PROP_VAL
Configuration property default vakue with a default outgoing packet filter
loaded by Tigase server.
|
static String |
SCHEDULER_THREADS_PROP_KEY
Configuration property key for setting number of threads used by component
ScheduledExecutorService.
|
protected static long |
SECOND
Constant used in time calculation procedures.
|
admins, ALL_PROP_KEY, COMMAND_PROP_NODE, scriptCommands, SCRIPTS_DIR_PROP_DEF, SCRIPTS_DIR_PROP_KEY, vHostManager
ADMINS_PROP_KEY, AMP_CLASS_NAME, AMP_CLUST_CLASS_NAME, AUTH_DOMAIN_POOL_CLASS, AUTH_REPO_DOMAINS_PROP_KEY, AUTH_REPO_PARAMS_NODE, AUTH_REPO_POOL_CLASS, AUTH_REPO_URL_PROP_KEY, BOSH_CLUST_COMP_CLASS_NAME, BOSH_COMP_CLASS_NAME, C2S_CLUST_COMP_CLASS_NAME, C2S_COMP_CLASS_NAME, CL_COMP_CLASS_NAME, CLUSTER_CONECT, CLUSTER_CONTR_CLASS_NAME, CLUSTER_LISTEN, CLUSTER_MODE, CLUSTER_NODES, CLUSTER_NODES_PROP_KEY, COMP_PROT_CLASS_NAME, COMPONENT_ID_PROP_KEY, DEF_AMP_NAME, DEF_BOSH_NAME, DEF_C2S_NAME, DEF_CL_COMP_NAME, DEF_CLUST_CONTR_NAME, DEF_COMP_PROT_NAME, DEF_EXT_COMP_NAME, DEF_HOSTNAME_PROP_KEY, DEF_MONITOR_NAME, DEF_S2S_NAME, DEF_SM_NAME, DEF_SRECV_NAME, DEF_SSEND_NAME, DEF_STATS_NAME, DEF_VHOST_MAN_NAME, DEF_WS2S_NAME, DERBY_REPO_CLASS_PROP_VAL, DERBY_REPO_URL_PROP_VAL, DRUPAL_REPO_URL_PROP_VAL, DRUPALWP_REPO_CLASS_PROP_VAL, DUMMY_REPO_CLASS_PROP_VAL, EXT_COMP_CLASS_NAME, GEN_ADMINS, GEN_AUTH_DB, GEN_AUTH_DB_URI, GEN_COMP_CLASS, GEN_COMP_NAME, GEN_CONF, GEN_CONFIG, GEN_CONFIG_ALL, GEN_CONFIG_COMP, GEN_CONFIG_CS, GEN_CONFIG_DEF, GEN_CONFIG_SM, GEN_DEBUG, GEN_DEBUG_PACKAGES, GEN_EXT_COMP, GEN_MAX_QUEUE_SIZE, GEN_SCRIPT_DIR, GEN_SM_PLUGINS, GEN_SREC_ADMINS, GEN_SREC_DB, GEN_SREC_DB_URI, GEN_TEST, GEN_TRUSTED, GEN_USER_DB, GEN_USER_DB_URI, GEN_USER_DB_URI_PROP_KEY, GEN_VIRT_HOSTS, HOSTNAMES_PROP_KEY, LIBRESOURCE_REPO_CLASS_PROP_VAL, LIBRESOURCE_REPO_URL_PROP_VAL, MONITOR_CLASS_NAME, MONITOR_CLUST_CLASS_NAME, MONITORING, MYSQL_REPO_CLASS_PROP_VAL, MYSQL_REPO_URL_PROP_VAL, PGSQL_REPO_CLASS_PROP_VAL, PGSQL_REPO_URL_PROP_VAL, ROUTER_COMP_CLASS_NAME, S2S_CLUST_COMP_CLASS_NAME, S2S_COMP_CLASS_NAME, SHARED_AUTH_REPO_PARAMS_PROP_KEY, SHARED_AUTH_REPO_PROP_KEY, SHARED_USER_REPO_PARAMS_PROP_KEY, SHARED_USER_REPO_PROP_KEY, SM_CLUST_COMP_CLASS_NAME, SM_COMP_CLASS_NAME, SRECV_COMP_CLASS_NAME, SSEND_COMP_CLASS_NAME, STANZA_WHITE_CHAR_ACK, STANZA_XMPP_ACK, STATS_CLASS_NAME, STRINGPREP_PROCESSOR, TIGASE_AUTH_REPO_CLASS_PROP_VAL, TIGASE_AUTH_REPO_URL_PROP_VAL, TIGASE_CUSTOM_AUTH_REPO_CLASS_PROP_VAL, TRUSTED_PROP_KEY, USER_DOMAIN_POOL_CLASS, USER_REPO_DOMAINS_PROP_KEY, USER_REPO_PARAMS_NODE, USER_REPO_POOL_CLASS, USER_REPO_POOL_SIZE, USER_REPO_POOL_SIZE_PROP_KEY, USER_REPO_URL_PROP_KEY, VHOST_MAN_CLASS_NAME, VHOST_MAN_CLUST_CLASS_NAME, WS2S_CLASS_NAME, WS2S_CLUST_CLASS_NAME, XML_REPO_CLASS_PROP_VAL, XML_REPO_URL_PROP_VAL, XMPP_STANZA_ACK
CMD_FEATURES, DEF_FEATURES, INFO_XMLNS, ITEMS_XMLNS
Constructor and Description |
---|
AbstractMessageReceiver() |
Modifier and Type | Method and Description |
---|---|
protected boolean |
addOutPacket(Packet packet)
Method description
|
protected boolean |
addOutPacketNB(Packet packet)
Non blocking version of
addOutPacket . |
protected boolean |
addOutPackets(Queue<Packet> packets)
Method description
|
boolean |
addOutPacketWithTimeout(Packet packet,
ReceiverTimeoutHandler handler,
long delay,
TimeUnit unit)
Method description
|
boolean |
addPacket(Packet packet)
Method adds a
Packet object to the internal input queue. |
boolean |
addPacketNB(Packet packet)
This is a variant of
addPacket(Packet) method which adds
Packet to in the internal input queue without blocking. |
boolean |
addPackets(Queue<Packet> packets)
This is a convenience method for adding all packets stored in given queue
to the component's internal input queue.
The method calls addPacket(Packet) in a loop for each packet
in the queue. |
void |
addRegexRouting(String address)
Method adds a new routing address for the component.
|
void |
addTimerTask(TimerTask task,
long delay)
Method queues and executes timer tasks using ScheduledExecutorService
which allows using more than one thread for executing tasks.
|
void |
addTimerTask(TimerTask task,
long delay)
Deprecated.
|
protected void |
addTimerTask(TimerTask task,
long delay,
TimeUnit unit)
Method queues and executes timer tasks using ScheduledExecutorService
which allows using more than one thread for executing tasks.
|
protected void |
addTimerTask(TimerTask task,
long delay,
TimeUnit unit)
Deprecated.
|
void |
clearRegexRoutings()
Method clears, removes all the component routing addresses.
|
void |
everyHour()
Utility method executed precisely every hour.
|
void |
everyMinute()
Utility method executed precisely every minute.
|
void |
everySecond()
Utility method executed precisely every second.
|
Map<String,Object> |
getDefaults(Map<String,Object> params)
Returns default configuration settings for the component as a
Map with keys as configuration property IDs and values as the
configuration property values. |
protected Integer |
getMaxQueueSize(int def)
Method description
|
Set<Pattern> |
getRegexRoutings()
Method returns a
Set with all component's routings as a
compiled regular expression patterns. |
void |
getStatistics(StatisticsList list)
Method returns component statistics.
|
int |
hashCodeForPacket(Packet packet)
This method decides how incoming packets are distributed among processing
threads.
|
boolean |
isInRegexRoutings(String address)
Method description
|
String |
newPacketId(String prefix)
Method description
|
int |
processingInThreads()
Concurrency control method.
|
int |
processingOutThreads()
Concurrency control method.
|
int |
processingThreads()
Deprecated.
|
void |
processOutPacket(Packet packet)
By default this method just copies the given packet between queue.
|
abstract void |
processPacket(Packet packet)
This is the main
Packet processing method. |
void |
processPacket(Packet packet,
Queue<Packet> results)
Method description
|
void |
release()
Method description
|
boolean |
removeRegexRouting(String address)
Method description
|
int |
schedulerThreads()
Method returns default number of threads used by SchedulerExecutorService
|
void |
setMaxQueueSize(int maxQueueSize)
Method description
|
void |
setName(String name)
Method description
|
void |
setParent(MessageReceiver parent)
Method description
|
void |
setProperties(Map<String,Object> props)
Sets all configuration properties for object.
|
void |
start()
Method description
|
void |
stop()
Method description
|
addComponentDomain, canCallCommand, checkCommandAcl, getComponentId, getComponentInfo, getDefHostName, getDefVHostItem, getDiscoCategory, getDiscoCategoryType, getDiscoDescription, getDiscoFeatures, getDiscoFeatures, getDiscoInfo, getDiscoInfo, getDiscoItems, getDiscoItems, getName, getScriptCommands, getScriptItems, getServiceEntity, getVHostItem, handlesLocalDomains, handlesNameSubdomains, handlesNonLocalDomains, initBindings, initializationCompleted, isAdmin, isInitializationComplete, isLocalDomain, isLocalDomainOrComponent, isNonAdminCommands, isSubdomain, processScriptCommand, removeComponentDomain, removeServiceDiscoveryItem, setVHostManager, updateServiceDiscoveryItem, updateServiceDiscoveryItem, updateServiceDiscoveryItem, updateServiceEntity
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
getDefHostName
getComponentId, getComponentInfo, getName, initializationCompleted, isInitializationComplete
public static final String INCOMING_FILTERS_PROP_KEY
public static final String INCOMING_FILTERS_PROP_VAL
PacketFilterIfc
interface.public static final String MAX_QUEUE_SIZE_PROP_KEY
public static final Integer MAX_QUEUE_SIZE_PROP_VAL
Runtime.getRuntime().maxMemory() / 400000L
You can change the
default queue size by setting a different value for the
MAX_QUEUE_SIZE_PROP_KEY
property in the server configuration.public static final String OUTGOING_FILTERS_PROP_KEY
PacketFilterIfc
interface.public static final String OUTGOING_FILTERS_PROP_VAL
PacketFilterIfc
interface.public static final String SCHEDULER_THREADS_PROP_KEY
protected static final long SECOND
protected static final long MINUTE
SECOND
s.protected static final long HOUR
MINUTE
s.protected int maxQueueSize
protected int maxInQueueSize
protected int maxOutQueueSize
public boolean addPacket(Packet packet)
Packet
object to the internal input queue.
Packets from the input queue are later passed to the
processPacket(Packet)
method. This is a blocking method
waiting if necessary for the room if the queue is full.
The method returns a boolean
value of true
if the
packet has been successfully added to the queue and false
otherwise.
There can be many queues and many threads processing packets for the
component, however the method makes the best effort to guarantee that
packets are later processed in the correct order. For example that packets
for a single user always end up in the same exact queue. You can tweak the
packets distribution among threads by overwriting
hashCodeForPacket(Packet)
method.N
threads the packets are distributed among thread
using following logic:
int threadNo = Math.abs(hashCodeForPacket(packet) % N);This is a preferred method to be used by most Tigase components. If the queues are full the component should stop and wait for more room. The blocking methods aim to prevent from the system overloading or wasting resources for generating packets which can't be processed anyway.
addPacket
in interface MessageReceiver
packet
- is a Packet
instance being put to the component
internal input queue.boolean
value of true
if the packet has
been successfully added to the queue and false
otherwise.public boolean addPacketNB(Packet packet)
addPacket(Packet)
method which adds
Packet
to in the internal input queue without blocking.
The method returns a boolean
value of true
if the
packet has been successful added to the queue and false
otherwise.
Use of the non-blocking methods is not recommended for most of the
components implementations. The only component which is allowed to use them
is the server MessageRouter
implementation which can not hang
on any method. This would cause a dead-lock in the application. All other
components must use blocking methods and wait if the system is under so
high load that it's queues are full.
See addPacket(Packet)
method's documentation for some more
details.addPacketNB
in interface MessageReceiver
packet
- is a Packet
instance being put to the component
internal input queue.boolean
value of true
if the packet has
been successfully added to the queue and false
otherwise.addPacket(Packet packet)
public boolean addPackets(Queue<Packet> packets)
addPacket(Packet)
in a loop for each packet
in the queue. If the call returns true
then the packet is
removed from the given queue, otherwise the methods ends the loop and
returns false
.
Please note, if the method returns true
it means that all the
packets from the queue passed as a parameter have been successfuly run
through the addPacket(Packet)
method and the queue passed as a
parameter should be empty. If the method returns false then at least one
packet from the parameter queue wasn't successfully run through the
addPacket(Packet)
method. If the method returns
false
then the queue passed as a parameter is not empty and it
contains packet which was unseccessfully run through the
addPacket(Packet)
method and all the packets which were not
run at all.addPackets
in interface MessageReceiver
packets
- is a Queue
of packets for adding to the component
internal input queue. All the packets are later processed by
processPacket(Packet)
method in the same exact order
if they are processed by the same thread. See documentation
hashCodeForPacket(Packet)
method how to control
assiging packets to particular threads.boolean
value of true
if all packets
has been successfully added to the component's internal input queue
and false
otherwise.hashCodeForPacket(Packet packet)
public void addRegexRouting(String address)
MessageRouter
to calculate packet's destination.
If the packet's destination address matches one of the component's routing
addresses the packet is added to the component's internal input queue.
By default all components accept packets addressed to the componentId and
to:
component.getName() + '@' + any virtual domainTODO: The future implementation most likely accept packets addressed to:
any virtual domain + '/' + component.getName()instead. The routings are passed as Java regular expression strings are the extra addresses accepted by the component. In most cases this is used by the external component protocol implementations which can dynamically change accepted addresses depending on the connected external components.
address
- is a Java regular expression string for the packet's destination
address accepted by this component.public void addTimerTask(TimerTask task, long delay)
task
- delay
- @Deprecated public void addTimerTask(TimerTask task, long delay)
task
- delay
- public void clearRegexRoutings()
public void everyHour()
public void everyMinute()
public void everySecond()
public int hashCodeForPacket(Packet packet)
int threadNo = Math.abs(hashCodeForPacket(packet) % N);For a PubSub component, for example, a better packets distribution would be based on the PubSub channel name, for SM a better distribution is based on the destination address, etc....
packet
- is a Packet
which needs to be processed by some
thread.public String newPacketId(String prefix)
prefix
- String
public int processingInThreads()
public int processingOutThreads()
@Deprecated public int processingThreads()
processingOutThreads()
and
processingInThreads()
instead.int
public void processOutPacket(Packet packet)
packet
- is an output packet which normally has to go to other component
for further processing.public abstract void processPacket(Packet packet)
Packet
processing method. It is called
concurrently from many threads so implementing it in thread save manner is
essential. The method is called for each packet addressed to the component.
Please note, the Packet
instance may be processed by different
parts of the server, different components or plugins at the same time.
Therefore this is very important to tread the Packet
instance
as unmodifiable object.
Processing in this method is asynchronous, therefore there is no result
value. If there are some 'result' packets generated during processing, they
should be passed back using addOutPacket(Packet)
method.packet
- is an instance of the Packet
class passed for
processing.public final void processPacket(Packet packet, Queue<Packet> results)
processPacket
in interface ServerComponent
processPacket
in class BasicComponent
packet
- results
- public void release()
release
in interface ServerComponent
release
in class BasicComponent
public boolean removeRegexRouting(String address)
address
- boolean
public int schedulerThreads()
int
public void start()
start
in interface MessageReceiver
public void stop()
public Map<String,Object> getDefaults(Map<String,Object> params)
Map
with keys as configuration property IDs and values as the
configuration property values. All the default parameters returned from
this method are later passed to the setProperties(...)
method.
Some of them may have changed value if they have been overwritten in the
server configuration. The configuration property value can be of any of the
basic types: int
, long
, boolean
,
String
.getDefaults
in interface Configurable
getDefaults
in class BasicComponent
params
- is a Map
with some initial properties set for the
starting up server. These parameters can be used as a hints to
generate component's default configuration.Map
with the component default configuration.public Set<Pattern> getRegexRoutings()
Set
with all component's routings as a
compiled regular expression patterns. The Set
can be empty but
it can not be null.Set
with all component's routings as a compiled
regular expression patterns.public void getStatistics(StatisticsList list)
Level.FINEST
assigned and must be put inside the level guard
to prevent generating them by the system monitor. The system monitor does
not collect FINEST
statistics.
Level guard code looks like the example below:
if (list.checkLevel(Level.FINEST)) { // Some CPU intensive calculations or lengthy operations list.add(getName(), "Statistic description", stat_value, Level.FINEST); }This way you make sure your extensive operation is not executed every second by the monitoring system and does not affect the server performance.
getStatistics
in interface StatisticsContainer
list
- is a StatistcsList
where all statistics are stored.public boolean isInRegexRoutings(String address)
isInRegexRoutings
in interface MessageReceiver
address
- boolean
public void setMaxQueueSize(int maxQueueSize)
maxQueueSize
- public void setName(String name)
setName
in interface ServerComponent
setName
in class BasicComponent
name
- public void setParent(MessageReceiver parent)
setParent
in interface MessageReceiver
parent
- @TODO(note="Replace fixed filers loading with configurable options for that") public void setProperties(Map<String,Object> props) throws ConfigurationException
setProperties
in interface Configurable
setProperties
in class BasicComponent
props
- ConfigurationException
protected boolean addOutPacket(Packet packet)
packet
- boolean
protected boolean addOutPacketNB(Packet packet)
addOutPacket
.packet
- a Packet
valueboolean
valueprotected boolean addOutPackets(Queue<Packet> packets)
packets
- boolean
public boolean addOutPacketWithTimeout(Packet packet, ReceiverTimeoutHandler handler, long delay, TimeUnit unit)
packet
- handler
- delay
- unit
- boolean
protected void addTimerTask(TimerTask task, long delay, TimeUnit unit)
task
- delay
- unit
- @Deprecated protected void addTimerTask(TimerTask task, long delay, TimeUnit unit)
task
- delay
- unit
- protected Integer getMaxQueueSize(int def)
def
- Integer
Copyright © 2014 "Tigase, Inc.". All rights reserved.