Skip to content

Instantly share code, notes, and snippets.

@danmueller

danmueller/.gitignore

Created March 25, 2009 11:31
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save danmueller/85430 to your computer and use it in GitHub Desktop.
Save danmueller/85430 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
.classpath
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" output="target/classes" path="src/main/java"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>
Raw
.gitconfig
[alias]
st = status
ci = commit
co = checkout
w = whatchanged
[diff]
renamelimit = 0
[color]
diff = auto
interactive = auto
status = auto
branch = auto
[apply]
whitespace = fix
Raw
.gitignore
*.tmproj
logs
Raw
.gitignore
bin
*eggs/
*.egg-info/
.installed.cfg
Raw
.gitignore
bin
*eggs/
*.egg-info/
.installed.cfg
Raw
.gitignore
build.xml
conf
perfharness
ruby
transactions
Raw
.paths
#!/bin/bash
# this file should be sourced (preferable with set -u)
export DOWNLOADS=$HOME/downloads
export PATH=$PATH:~/.scripts/bin
#EC2_BASE=~/.ec2
#export EC2_HOME=`ls -1td $EC2_BASE/ec2-api-tools* | head -n1`
#export PATH=$PATH:/opt/local/bin:/opt/local/sbin:$EC2_HOME/bin:~/.scripts/bin
#export EC2_PRIVATE_KEY=`ls $EC2_BASE/certs/pk-*.pem`
#export EC2_CERT=`ls $EC2_BASE/certs/cert-*.pem`
Raw
.profile
#!/bin/bash
source ~/.paths
bind '"\M-d": backward-kill-word'
pwd=`which pwd`
alias ls='ls --color'
function ll () {
ls -hl $1
$pwd
}
function la () {
ls -hla $1
$pwd
}
alias l=ll
#alias cd="pushd > /dev/null"
alias bd="popd"
alias dirs="dirs -v"
function e {
if [ -z $1 ]; then
cd `cat ~/.login_directory`
else
pushd . > /dev/null
cd $1
pwd > ~/.login_directory
popd > /dev/null
fi
}
function e? {
cat ~/.login_directory
}
Raw
.project
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>jms-utils</name>
<comment></comment>
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.jdt.core.javabuilder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.maven.ide.eclipse.maven2Builder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
</natures>
</projectDescription>
Raw
.screenrc
vbell off
escape ``
startup_message off
hardstatus on
hardstatus alwayslastline
hardstatus string "%{.bW}%-w%{.rW}%n %t%{-}%+w %=%{..G} %H %{..Y} %m/%d %C%a "
# ccg: the following copied from geoflop:~sysadmin/.screenrc
#
## Example of a user's .screenrc file
##
#
## http://www.orkut.com/CommMsgs.aspx?cmm=1988&tid=4
#
## This is how one can set a reattach password:
## password ODSJQf.4IJN7E # "1234"
#
## no annoying audible bell, please
#vbell on
#
## detach on hangup
#autodetach on
#
## don't display the copyright page
#startup_message off
#
## emulate .logout message
#pow_detach_msg "Screen session of \$LOGNAME \$:cr:\$:nl:ended."
#
## advertise hardstatus support to $TERMCAP
## termcapinfo * '' 'hs:ts=\E_:fs=\E\\:ds=\E_\E\\'
#
## make the shell in every window a login shell
shell -$SHELL
#
## autoaka testing
## shellaka '> |tcsh'
## shellaka '$ |sh'
#
## set every new windows hardstatus line to somenthing descriptive
## defhstatus "screen: ^En (^Et)"
#
#defscrollback 100000
#
## don't kill window after the process died
## zombie "^["
#
#################
##
## xterm tweaks
##
#
##xterm understands both im/ic and doesn't have a status line.
##Note: Do not specify im and ic in the real termcap/info file as
##some programs (e.g. vi) will not work anymore.
termcap xterm-color hs@:cs=\E[%i%d;%dr:im=\E[4h:ei=\E[4l
terminfo xterm-color hs@:cs=\E[%i%p1%d;%p2%dr:im=\E[4h:ei=\E[4l
#
##80/132 column switching must be enabled for ^AW to work
##change init sequence to not switch width
termcapinfo xterm-color Z0=\E[?3h:Z1=\E[?3l:is=\E[r\E[m\E[2J\E[H\E[?7h\E[?1;4;6l
#
## Make the output buffer large for (fast) xterms.
termcapinfo xterm* OL=10000
#
## tell screen that xterm can switch to dark background and has function
## keys.
termcapinfo xterm-color 'VR=\E[?5h:VN=\E[?5l'
termcapinfo xterm-color 'k1=\E[11~:k2=\E[12~:k3=\E[13~:k4=\E[14~'
termcapinfo xterm-color 'kh=\E[1~:kI=\E[2~:kD=\E[3~:kH=\E[4~:kP=\E[H:kN=\E[6~'
#
## special xterm hardstatus: use the window title.
termcapinfo xterm-color 'hs:ts=\E]2;:fs=\007:ds=\E]2;screen\007'
termcapinfo xterm-color 'hs:ts=\E]2;:fs=\007:ds=\E]1;screen\007'
#
terminfo xterm-color 'vb=\E[?5h$<200/>\E[?5l'
termcapinfo xterm-color 'vi=\E[?25l:ve=\E[34h\E[?25h:vs=\E[34l'
#
## emulate part of the 'K' charset
#termcapinfo xterm-color 'XC=K%,%\E(B,[\304,\\\\\326,]\334,{\344,|\366,}\374,~\337'
#
## xterm-52 tweaks:
## - uses background color for delete operations
#termcapinfo xterm ut
#
## pgup pgdn for xterm
#termcapinfo xterm*|rxvt* 'ti@:te@'
#
#################
##
## wyse terminals
##
#
##wyse-75-42 must have flow control (xo = "terminal uses xon/xoff")
##essential to have it here, as this is a slow terminal.
#termcapinfo wy75-42 xo:hs@
#
## New termcap sequences for cursor application mode.
#termcapinfo wy* CS=\E[?1h:CE=\E[?1l:vi=\E[?25l:ve=\E[?25h:VR=\E[?5h:VN=\E[?5l:cb=\E[1K:CD=\E[1J
#
#################
##
## other terminals
##
#
##make hp700 termcap/info better
#termcapinfo hp700 'Z0=\E[?3h:Z1=\E[?3l:hs:ts=\E[62"p\E[0$~\E[2$~\E[1$}:fs=\E[0}\E[61"p:ds=\E[62"p\E[1$~\E[61"p:ic@'
#
## Extend the vt100 desciption by some sequences.
#termcap vt100* ms:AL=\E[%dL:DL=\E[%dM:UP=\E[%dA:DO=\E[%dB:LE=\E[%dD:RI=\E[%dC
#terminfo vt100* ms:AL=\E[%p1%dL:DL=\E[%p1%dM:UP=\E[%p1%dA:DO=\E[%p1%dB:LE=\E[%p1%dD:RI=\E[%p1%dC
#
#
#################
##
## 256 colours
##
## ... http://frexx.de/xterm-256-notes/
##
#
## terminfo and termcap for nice 256 color terminal
## allow bold colors - necessary for some reason
#attrcolor b ".I"
## tell screen how to set colors. AB = background, AF=foreground
#termcapinfo xterm 'Co#256:AB=\E[48;5;%dm:AF=\E[38;5;%dm'
## erase background with current bg color
#defbce "on"
#
##term "screen-256color"
#
#
#################
##
## ssh-agent support
##
## http://www.jukie.net/~bart/blog/screen-ssh-agent
##
#
#unsetenv SSH_AUTH_SOCK
#setenv SSH_AUTH_SOCK "$HOME/.screen/ssh-auth-sock.$HOSTNAME"
#
#################
##
## keybindings
##
#
##remove some stupid / dangerous key bindings
#bind k
#bind ^k
#bind .
#bind ^\
#bind \\
#bind ^h
#bind h
##make them better
#bind 'K' kill
#bind 'I' login on
#bind 'O' login off
#bind '}' history
#
## Yet another hack:
## Prepend/append register [/] to the paste if ^a^] is pressed.
## This lets me have autoindent mode in vi.
#register [ "\033:se noai\015a"
#register ] "\033:se ai\015a"
#bind ^] paste [.]
#
#################
##
## default windows
##
#
## screen -t local 0
## screen -t mail 1 elm
## screen -t 40 2 rlogin faui40
#
## caption always "%3n %t%? @%u%?%? [%h]%?"
## hardstatus alwaysignore
## hardstatus alwayslastline "%w"
#
## caption always "%{Yb} %Y-%02m-%02d %0c %{k}|%{C} %LD %{k}|%{G} %l %{k}|%{W} %-Lw%{+u}%n %t%{-u}%+Lw"
#
#caption always "%{Yb} %D %Y-%02m-%02d %0c %{k}|%{G} %l %{k}|%{W} %-w%{+b}%n %t%{-b}%+w"
#
Raw
__init__.py
import logging
logging.basicConfig(level=logging.INFO,
format='[%(name)s %(levelname)s] %(message)s')
# format='[%(levelname)s] %(message)s')
logging.getLogger('boto').setLevel(logging.DEBUG)
Raw
__init__.py
Raw
README
This collection of scripts was written while I was working on a stress test
for ActiveMQ running on Amazon EC2.
If you find anything of value, go ahead and make use of it. If you one day
discover that you have something that might be of value to someone else,
share it.
Raw
README
some helper shell scripts to help with admin tasks
and machines running under EC2.
Raw
README
This is a modified version of the sources found in the
$ACTIVEMQ_HOME/examples
directory of the activemq 5.2.0 bin download. The original
sources are licensed under the Apache License v2.0. The
modified sources here are released under the same license.
The modifications are minor as you can see when doing a diff
between the actual source directory and the orig directory.
Raw
activemq-example-tool.txt
[echo]
[echo] This script requires Ant 1.6 or higher
[echo]
[echo] NOTE: All options should be specified as system properties
[echo] on the command line, e.g.:
[echo]
[echo] ant consumer -Durl=tcp://hostname:1234 -Dtopic=true
[echo]
[echo] OR
[echo]
[echo] ant producer -Durl=tcp://hostname:61616 -Dmax=1000
[echo]
[echo] Usage:
[echo] ant <ant-task> <options>
[echo]
[echo] --------------------------------------------------------
[echo] ant consumer <options> - Creates a consumer which waits until a specific number
[echo] of messages have been received
[echo]
[echo] Consumer Options:
[echo] url - Used to specify acustom URL for the
[echo] broker, e.g., tcp://hostname:1234
[echo] topic - A boolean to determine whether to use
[echo] topics or queues; the default is false
[echo] subject - Used to specify a custom destination
[echo] name, e.g. MyDestination
[echo] durable - A boolean to specify that you want to
[echo] create a durable topic?
[echo] max - The maximum number of messages to wait
[echo] for before shutting down
[echo] clientId - A string to use as the client id
[echo] transacted - A boolean to specify that you want to use
[echo] transactions?
[echo] sleepTime - The time to sleep between message consumptions
[echo] verbose - Used to print out more info; the default is
[echo] true
[echo] ack-mode - The type of message acknowledgement to use;
[echo] see the Javadocs for javax.jms.Session for
[echo] more information
[echo] receive-time-out - An integer to specify the time to wait for
[echo] message consumption
[echo]
[echo]
[echo] --------------------------------------------------------
[echo] ant producer <options> - Creates a producer publishing a number of messages
[echo]
[echo] Producer Options:
[echo] url - Used to specify acustom URL for the broker,
[echo] e.g., tcp://hostname:1234
[echo] topic - A boolean to determine whether to use topics
[echo] or queues
[echo] subject - Used to specify a custom destination name,
[echo] e.g. MyDestination
[echo] durable - A boolean to specify that you want to create
[echo] a durable topic subscriber?
[echo] max - The maximum number of messages to wait for
[echo] before shutting down
[echo] sleepTime - The time to sleep between message consumptions
[echo] transacted - A boolean to specify that you want to use
[echo] transactions?
[echo] verbose - Used to print out more info; the default is true
[echo]
[echo] --------------------------------------------------------
[echo]
[echo] ant -help - Display ant help screen
[echo] ant help - Display this message
[echo] ant clean - Delete the built directory
[echo] ant embedBroker - Runs an embedded broker inside Java code
[echo] ant war - Creates a WAR deployment unit of the ActiveMQ Broker
[echo]
Raw
activemq-orig.xml
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:///${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-a" dataDirectory="${activemq.base}/data">
<!-- Use the following to configure how ActiveMQ is exposed in JMX -->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers -->
<networkConnector name="default-nc" uri="multicast://default"/>
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
</networkConnectors>
<persistenceAdapter>
<amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/data" maxFileLength="20 mb"/>
</persistenceAdapter>
<!-- Use the following if you wish to configure the journal with JDBC -->
<!--
<persistenceAdapter>
<journaledJDBC dataDirectory="${activemq.base}/data" dataSource="#postgres-ds"/>
</persistenceAdapter>
-->
<!-- Or if you want to use pure JDBC without a journal -->
<!--
<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#postgres-ds"/>
</persistenceAdapter>
-->
<sslContext>
<sslContext keyStore="file:${activemq.base}/conf/broker.ks" keyStorePassword="password" trustStore="file:${activemq.base}/conf/broker.ts" trustStorePassword="password"/>
</sslContext>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="1 gb" name="foo"/>
</storeUsage>
<tempUsage>
<tempUsage limit="500 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<!-- The transport connectors ActiveMQ will listen to -->
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61616" discoveryUri="multicast://default"/>
</transportConnectors>
</broker>
<!-- Uncomment to create a command agent to respond to message based admin commands on the ActiveMQ.Agent topic -->
<!--
<commandAgent xmlns="http://activemq.apache.org/schema/core" brokerUrl="vm://localhost" username="${activemq.username}" password="${activemq.password}"/>
-->
<!-- An embedded servlet engine for serving up the Admin console -->
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8161"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>
<!-- This xbean configuration file supports all the standard spring xml configuration options -->
<!-- Postgres DataSource Sample Setup -->
<!--
<bean id="postgres-ds" class="org.postgresql.ds.PGPoolingDataSource">
<property name="serverName" value="localhost"/>
<property name="databaseName" value="activemq"/>
<property name="portNumber" value="0"/>
<property name="user" value="activemq"/>
<property name="password" value="activemq"/>
<property name="dataSourceName" value="postgres"/>
<property name="initialConnections" value="1"/>
<property name="maxConnections" value="10"/>
</bean>
-->
<!-- MySql DataSource Sample Setup -->
<!--
<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
<property name="username" value="activemq"/>
<property name="password" value="activemq"/>
<property name="maxActive" value="200"/>
<property name="poolPreparedStatements" value="true"/>
</bean>
-->
<!-- Oracle DataSource Sample Setup -->
<!--
<bean id="oracle-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="oracle.jdbc.driver.OracleDriver"/>
<property name="url" value="jdbc:oracle:thin:@localhost:1521:AMQDB"/>
<property name="username" value="scott"/>
<property name="password" value="tiger"/>
<property name="maxActive" value="200"/>
<property name="poolPreparedStatements" value="true"/>
</bean>
-->
<!-- Embedded Derby DataSource Sample Setup -->
<!--
<bean id="derby-ds" class="org.apache.derby.jdbc.EmbeddedDataSource">
<property name="databaseName" value="derbydb"/>
<property name="createDatabase" value="create"/>
</bean>
-->
</beans>
Raw
activemq-start
#!/bin/bash
$HOME/apache-activemq-5.2.0/bin/activemq &
Raw
activemq-stop
#!/bin/bash
kill `ps -ef | grep apache-activemq-5\.2\.0 | grep -v grep | grep -v kill | awk '{print $2}'`
Raw
activemq.py
#!/usr/bin/env python
"""
This file is used to actually start the broker on the target system and have it behave.
"""
import os
import sys
import re
import subprocess
import logging
import optparse
import types
#from instances import aws_connect
logging.getLogger('boto').setLevel(logging.INFO)
log = logging.getLogger('activemq_stress')
def main():
usage = "usage: %prog [options]"
parser = optparse.OptionParser(usage)
parser.add_option("-u", "--up", dest="start",
action="store_true",
default=False,
help="Start the broker.")
parser.add_option("-d", "--down", dest="stop",
action="store_true",
default=False,
help="Stop the broker.")
parser.add_option("-t", "--status", dest="status",
action="store_true",
default=False,
help="Display status message for the broker")
parser.add_option("-c", "--connection-url", dest="connection_url",
action="store_true",
default=False,
help="Display connection URL for clients to use all active brokers in group [activemq-broker]")
parser.add_option("-l", "--local", dest="local",
action="store_false",
default=True,
help="all operations are only executed locally, and fail if there's remote connectivity required")
options, args = parser.parse_args()
if options.start and options.stop:
raise Exception, 'Cannot start and stop at the same time, decide what you want'
# Merging branch requires latest merged master
if options.start:
start_broker(options.local)
elif options.stop:
stop_broker()
if options.status:
print_broker_status()
if options.connection_url:
if options.local:
raise Exception, 'online is required'
else:
conn = aws_connect()
reservation = conn.get_all_instances()
def start_broker(local=True):
# should we block, until it's running or waiting for the master?
log.info('starting activemq')
# check implicitly on running brokers and ask for restart
cmd = os.path.join(activemq_home, 'bin', 'activemq')
if os.name == 'nt':
cmd += '.bat'
if local:
cmd += " xbean:file:" + os.path.join(config_home, "inmemory-activemq.xml")
std_log = file(os.path.join(log_dir, 'activemq_std.log'), 'a')
err_log = file(os.path.join(log_dir, 'activemq_err.log'), 'a')
process = subprocess.Popen(cmd,
shell=True,
stdout=std_log,
stderr=err_log)
# output,_ = call_command("%s" % cmd)
def get_broker_pids():
cmd = "ps aux | grep %s | grep -v grep | awk '{print $2}'" % activemq_home
log.debug("'%s'" % cmd)
process = subprocess.Popen(cmd,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output,_ = process.communicate()
return [int(s) for s in output.splitlines()]
def print_broker_status():
cmd = "ps aux | grep %s | grep -v grep" % activemq_home
log.debug("'%s'" % cmd)
process = subprocess.Popen(cmd,
shell=True)
def stop_broker(pids=None):
# if called without argument, go on a killing spree!!
if not pids:
pids = get_broker_pids()
if type(pids) is types.ListType:
for pid in pids:
stop_broker(pid) # recursive
else:
log.info('[pid: %s] killing activemq' % pids)
log.debug("%s [%s]" % (pids, type(pids)))
os.kill(pids, 15) # 15 is SIG TERM, see man kill for numbers
def call_command(command):
log.warning("'%s'" % command)
#return (None, None)
if False:
process = subprocess.Popen(command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
process = subprocess.Popen(command, shell=True)
log.warning("'%s'" % command)
return (None, None)
return process.wait()
def setup_dir_from_env(var_name, default=None, create_if_absent=False):
var = os.getenv(var_name)
if not var:
if default:
log.warn("%s wasn't defined in the environment" % var_name)
var = default
else:
raise Exception, "%s not defined, and no default defined either" % var_name
# check if it's a decent path
if not os.path.exists(var):
if create_if_absent:
os.makedirs(var)
return var
else:
raise Exception, "%s is defined as %s which is not a directory" % (var_name, var)
if not os.path.isdir(var):
raise Exception, "%s is defined as %s which is not a directory" % (var_name, var)
# if not os.path.isreadable(var):
# raise Exception, "%s is defined as %s is not readable" % (var_name, var)
# if not os.path.iswriteable(var):
# raise Exception, "%s is defined as %s is not writeable" % (var_name, var)
return var
activemq_home = setup_dir_from_env('ACTIVEMQ_HOME',
default='~/apache-activemq-5.2.0-bin')
log_dir = setup_dir_from_env('STRESS_LOG_DIR',
default='~/test-logs',
create_if_absent=True)
script_path = os.path.join(sys.path[0], __name__.replace(".", "/")) # dir of the current executing script
script_home = os.path.dirname(script_path)
config_home = os.path.join(script_home, 'configs')
if __name__ == "__main__":
main()
Raw
activemq.xml
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:///${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost">
<!-- Use the following to configure how ActiveMQ is exposed in JMX -->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#mssql-jtds-ds" adapter="#sqlserver-adapter"/>
</persistenceAdapter>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<tempUsage>
<tempUsage limit="500 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61616"/>
</transportConnectors>
<!-- Ignored
<plugins>
<jaasAuthenticationPlugin configuration="activemq-domain" />
</plugins>
-->
</broker>
<!-- Uncomment to create a command agent to respond to
message based admin commands on the ActiveMQ.Agent topic -->
<!--
<commandAgent xmlns="http://activemq.apache.org/schema/core" brokerUrl="vm://localhost" username="${activemq.username}" password="${activemq.password}"/>
-->
<!-- An embedded servlet engine for serving up the Admin console -->
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8161"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>
<!-- This xbean configuration file supports all the standard spring xml configuration options -->
<!-- Postgres DataSource Sample Setup -->
<!--
<bean id="postgres-ds" class="org.postgresql.ds.PGPoolingDataSource">
<property name="serverName" value="localhost"/>
<property name="databaseName" value="activemq"/>
<property name="portNumber" value="0"/>
<property name="user" value="activemq"/>
<property name="password" value="activemq"/>
<property name="dataSourceName" value="postgres"/>
<property name="initialConnections" value="1"/>
<property name="maxConnections" value="10"/>
</bean>
-->
<!-- JTDS (DBCP) DataSource Sample Setup -->
<bean id="mssql-jtds-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="net.sourceforge.jtds.jdbcx.JtdsDataSource"/>
<property name="url" value="jdbc:jtds:sqlserver://HOST:1433;DatabaseName=ACTIVEMQ"/>
<property name="username" value="blabl"/>
<property name="password" value="blabla"/>
<property name="poolPreparedStatements" value="false"/>
<property name="maxWait" value="20"/>
<property name="maxIdle" value="10"/>
<property name="maxActive" value="20"/>
<property name="validationQuery" value="select cast('10' as integer)"/>
<property name="testOnBorrow" value="false"/>
<property name="defaultAutoCommit" value="false"/>
</bean>
<bean id="sqlserver-adapter" class="ch.bedag.ste.opensource.activemq.jdbc.adapter.SqlServerJDBCAdapter"/>
<bean id="ds-exporter" class="org.springframework.jmx.export.MBeanExporter" lazy-init="false">
<property name="beans">
<map>
<entry key="DataSource:name=mssql" value-ref="mssql-jtds-ds"/>
</map>
</property>
<property name="registrationBehaviorName" value="REGISTRATION_REPLACE_EXISTING"/>
<property name="assembler">
<bean class="org.springframework.jmx.export.assembler.MethodNameBasedMBeanInfoAssembler">
<property name="managedMethods">
<value>getDriverClassName,getDefaultAutoCommit,getLogAbandoned,setLogAbandoned,getMaxActive,getMaxIdle,getMaxOpenPreparedStatements,getMaxWait,getMinIdle,getNumActive,getNumIdle,isPoolPreparedStatements,getUrl,getUsername,getValidationQuery,getTestOnBorrow,getTestOnReturn,getTestWhileIdle</value>
</property>
</bean>
</property>
</bean>
</beans>
Raw
bedag-test-1a
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
if [ $# -eq 0 ]; then
program=`basename $0`
echo " $program <url>"
echo " <url> is of the form tcp://host:port"
echo " standard port is 61616"
exit 1
fi
TEST_NAME=producer-1a
STD_LOG=$BTLOGS_HOME/$TEST_NAME.log
ERR_LOG=$BTLOGS_HOME/$TEST_NAME-error.log
echo "$STD_LOG"
echo "$ERR_LOG"
time ant -f $BUILD_FILE \
producer \
-Durl=$1 \
-Dmax=1000000 \
-DmessageSize=10240 \
-Dtransacted=true \
> $STD_LOG \
&2> $ERR_LOG \
&
Raw
bedag-test-1b
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
if [ $# -eq 0 ]; then
program=`basename $0`
echo " $program <url>"
echo " <url> is of the form tcp://host:port"
echo " standard port is 61616"
exit 1
fi
TEST_NAME=producer-1b
STD_LOG=$BTLOGS_HOME/$TEST_NAME.log
ERR_LOG=$BTLOGS_HOME/$TEST_NAME-error.log
echo "$STD_LOG"
echo "$ERR_LOG"
BUILD_FILE=$ACTIVEMQ_HOME/example/build.xml
time ant -f $BUILD_FILE \
consumers \
-Durl=$1 \
-Dmax=1000000 \
-DmessageSize=10240 \
-Dtransacted=true \
> $STD_LOG \
&2> $ERR_LOG \
&
Raw
bedag-test-1c
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
if [ $# -eq 0 ]; then
program=`basename $0`
echo " $program <url>"
echo " <url> is of the form tcp://host:port"
echo " standard port is 61616"
exit 1
fi
TEST_NAME=producer-1a
STD_LOG=$BTLOGS_HOME/$TEST_NAME.log
ERR_LOG=$BTLOGS_HOME/$TEST_NAME-error.log
echo "$STD_LOG"
echo "$ERR_LOG"
for ((i=1;i<=4;i+=1));
#for i in {1..4};
do
time ant -f $BUILD_FILE \
producer \
-Durl=$1 \
-Dmax=1000000 \
-DmessageSize=1024 \
-Dtransacted=true \
> $STD_LOG$i \
&2> $ERR_LOG$i \
&
done
Raw
bedag-test-1d
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
if [ $# -eq 0 ]; then
program=`basename $0`
echo " $program <url>"
echo " <url> is of the form tcp://host:port"
echo " standard port is 61616"
exit 1
fi
TEST_NAME=producer-1a
STD_LOG=$BTLOGS_HOME/$TEST_NAME.log
ERR_LOG=$BTLOGS_HOME/$TEST_NAME-error.log
echo "$STD_LOG"
echo "$ERR_LOG"
for ((i=1;i<=4;i+=1));
#for i in {1..4};
do
time ant -f $BUILD_FILE \
consumer \
-Durl=$1 \
-receive-time-out=3600 \
-Dtransacted=true \
> $STD_LOG$i \
&2> $ERR_LOG$i \
&
done
Raw
bootstrap.py
##############################################################################
#
# Copyright (c) 2006 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Bootstrap a buildout-based project
Simply run this script in a directory containing a buildout.cfg.
The script accepts buildout command-line options, so you can
use the -c option to specify an alternate configuration file.
$Id$
"""
import os, shutil, sys, tempfile, urllib2
tmpeggs = tempfile.mkdtemp()
is_jython = sys.platform.startswith('java')
try:
import pkg_resources
except ImportError:
ez = {}
exec urllib2.urlopen('http://peak.telecommunity.com/dist/ez_setup.py'
).read() in ez
ez['use_setuptools'](to_dir=tmpeggs, download_delay=0)
import pkg_resources
if sys.platform == 'win32':
def quote(c):
if ' ' in c:
return '"%s"' % c # work around spawn lamosity on windows
else:
return c
else:
def quote (c):
return c
cmd = 'from setuptools.command.easy_install import main; main()'
ws = pkg_resources.working_set
if len(sys.argv) > 2 and sys.argv[1] == '--version':
VERSION = ' == %s' % sys.argv[2]
args = sys.argv[3:] + ['bootstrap']
else:
VERSION = ''
args = sys.argv[1:] + ['bootstrap']
if is_jython:
import subprocess
assert subprocess.Popen([sys.executable] + ['-c', quote(cmd), '-mqNxd',
quote(tmpeggs), 'zc.buildout' + VERSION],
env=dict(os.environ,
PYTHONPATH=
ws.find(pkg_resources.Requirement.parse('setuptools')).location
),
).wait() == 0
else:
assert os.spawnle(
os.P_WAIT, sys.executable, quote (sys.executable),
'-c', quote (cmd), '-mqNxd', quote (tmpeggs), 'zc.buildout' + VERSION,
dict(os.environ,
PYTHONPATH=
ws.find(pkg_resources.Requirement.parse('setuptools')).location
),
) == 0
ws.add_entry(tmpeggs)
ws.require('zc.buildout' + VERSION)
import zc.buildout.buildout
zc.buildout.buildout.main(args)
shutil.rmtree(tmpeggs)
Raw
bootstrap.py
##############################################################################
#
# Copyright (c) 2006 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Bootstrap a buildout-based project
Simply run this script in a directory containing a buildout.cfg.
The script accepts buildout command-line options, so you can
use the -c option to specify an alternate configuration file.
$Id$
"""
import os, shutil, sys, tempfile, urllib2
tmpeggs = tempfile.mkdtemp()
is_jython = sys.platform.startswith('java')
try:
import pkg_resources
except ImportError:
ez = {}
exec urllib2.urlopen('http://peak.telecommunity.com/dist/ez_setup.py'
).read() in ez
ez['use_setuptools'](to_dir=tmpeggs, download_delay=0)
import pkg_resources
if sys.platform == 'win32':
def quote(c):
if ' ' in c:
return '"%s"' % c # work around spawn lamosity on windows
else:
return c
else:
def quote (c):
return c
cmd = 'from setuptools.command.easy_install import main; main()'
ws = pkg_resources.working_set
if len(sys.argv) > 2 and sys.argv[1] == '--version':
VERSION = ' == %s' % sys.argv[2]
args = sys.argv[3:] + ['bootstrap']
else:
VERSION = ''
args = sys.argv[1:] + ['bootstrap']
if is_jython:
import subprocess
assert subprocess.Popen([sys.executable] + ['-c', quote(cmd), '-mqNxd',
quote(tmpeggs), 'zc.buildout' + VERSION],
env=dict(os.environ,
PYTHONPATH=
ws.find(pkg_resources.Requirement.parse('setuptools')).location
),
).wait() == 0
else:
assert os.spawnle(
os.P_WAIT, sys.executable, quote (sys.executable),
'-c', quote (cmd), '-mqNxd', quote (tmpeggs), 'zc.buildout' + VERSION,
dict(os.environ,
PYTHONPATH=
ws.find(pkg_resources.Requirement.parse('setuptools')).location
),
) == 0
ws.add_entry(tmpeggs)
ws.require('zc.buildout' + VERSION)
import zc.buildout.buildout
zc.buildout.buildout.main(args)
shutil.rmtree(tmpeggs)
Raw
broker-a-start
#!/bin/bash
~/broker-a/bin/activemq &
Raw
broker-a-stop
#!/bin/bash
kill `ps -ef | grep broker-a | grep -v grep | grep -v kill | awk '{print $2}'`
Raw
broker-a.xml
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:///${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-a" dataDirectory="${activemq.base}/data">
<!-- Use the following to configure how ActiveMQ is exposed in JMX -->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers
<networkConnector name="default-nc" uri="multicast://default"/>
-->
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
</networkConnectors>
<persistenceAdapter>
<amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/data" maxFileLength="20 mb"/>
</persistenceAdapter>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="1 gb" name="foo"/>
</storeUsage>
<tempUsage>
<tempUsage limit="500 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<!-- The transport connectors ActiveMQ will listen to -->
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61617" discoveryUri="multicast://default"/>
</transportConnectors>
</broker>
<!-- An embedded servlet engine for serving up the Admin console -->
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8162"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>
<!-- Embedded Derby DataSource Sample Setup -->
<!--
<bean id="derby-ds" class="org.apache.derby.jdbc.EmbeddedDataSource">
<property name="databaseName" value="derbydb"/>
<property name="createDatabase" value="create"/>
</bean>
-->
</beans>
Raw
broker-b-start
#!/bin/bash
~/broker-b/bin/activemq &
Raw
broker-b-stop
#!/bin/bash
kill `ps -ef | grep broker-b | grep -v grep | grep -v kill | awk '{print $2}'`
Raw
broker-b.xml
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:///${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-a" dataDirectory="${activemq.base}/data">
<!-- Use the following to configure how ActiveMQ is exposed in JMX -->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers
<networkConnector name="default-nc" uri="multicast://default"/>
-->
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
</networkConnectors>
<persistenceAdapter>
<amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/data" maxFileLength="20 mb"/>
</persistenceAdapter>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="1 gb" name="foo"/>
</storeUsage>
<tempUsage>
<tempUsage limit="500 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<!-- The transport connectors ActiveMQ will listen to -->
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61618" discoveryUri="multicast://default"/>
</transportConnectors>
</broker>
<!-- An embedded servlet engine for serving up the Admin console -->
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8163"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>
<!-- Embedded Derby DataSource Sample Setup -->
<!--
<bean id="derby-ds" class="org.apache.derby.jdbc.EmbeddedDataSource">
<property name="databaseName" value="derbydb"/>
<property name="createDatabase" value="create"/>
</bean>
-->
</beans>
Raw
buildout.cfg
[config]
mypkgs = activemq_stress
[buildout]
develop = .
parts = python scripts test
[python]
recipe = zc.recipe.egg
interpreter = python
eggs = ${config:mypkgs}
[scripts]
recipe = zc.recipe.egg:scripts
eggs = ${config:mypkgs}
[test]
recipe = zc.recipe.testrunner
eggs = ${config:mypkgs}
Raw
buildout.cfg
[config]
mypkgs = com.digitalstrider.tools.syspack
[buildout]
develop = .
parts = python scripts test
[python]
recipe = zc.recipe.egg
interpreter = python
eggs = ${config:mypkgs}
[scripts]
recipe = zc.recipe.egg:scripts
eggs = ${config:mypkgs}
[test]
recipe = zc.recipe.testrunner
eggs = ${config:mypkgs}
Raw
CommandLineSupport.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.ArrayList;
import org.apache.activemq.util.IntrospectionSupport;
/**
* Helper utility that can be used to set the properties on any object using
* command line arguments.
*
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
public final class CommandLineSupport {
private CommandLineSupport() {
}
/**
* Sets the properties of an object given the command line args.
*
* if args contains: --ack-mode=AUTO --url=tcp://localhost:61616 --persistent
*
* then it will try to call the following setters on the target object.
*
* target.setAckMode("AUTO");
* target.setURL(new URI("tcp://localhost:61616") );
* target.setPersistent(true);
*
* Notice the the proper conversion for the argument is determined by examining the
* setter arguement type.
*
* @param target the object that will have it's properties set
* @param args the commline options
* @return any arguments that are not valid options for the target
*/
public static String[] setOptions(Object target, String[] args) {
ArrayList<String> rc = new ArrayList<String>();
for (int i = 0; i < args.length; i++) {
if (args[i] == null) {
continue;
}
if (args[i].startsWith("--")) {
// --options without a specified value are considered boolean
// flags that are enabled.
String value = "true";
String name = args[i].substring(2);
// if --option=value case
int p = name.indexOf("=");
if (p > 0) {
value = name.substring(p + 1);
name = name.substring(0, p);
}
// name not set, then it's an unrecognized option
if (name.length() == 0) {
rc.add(args[i]);
continue;
}
String propName = convertOptionToPropertyName(name);
if (!IntrospectionSupport.setProperty(target, propName, value)) {
rc.add(args[i]);
continue;
}
}
}
String r[] = new String[rc.size()];
rc.toArray(r);
return r;
}
/**
* converts strings like: test-enabled to testEnabled
*
* @param name
* @return
*/
private static String convertOptionToPropertyName(String name) {
String rc = "";
// Look for '-' and strip and then convert the subsequent char to
// uppercase
int p = name.indexOf("-");
while (p > 0) {
// strip
rc += name.substring(0, p);
name = name.substring(p + 1);
// can I convert the next char to upper?
if (name.length() > 0) {
rc += name.substring(0, 1).toUpperCase();
name = name.substring(1);
}
p = name.indexOf("-");
}
return rc + name;
}
}
Raw
CommandLineSupport.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.ArrayList;
import org.apache.activemq.util.IntrospectionSupport;
/**
* Helper utility that can be used to set the properties on any object using
* command line arguments.
*
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
public final class CommandLineSupport {
private CommandLineSupport() {
}
/**
* Sets the properties of an object given the command line args.
*
* if args contains: --ack-mode=AUTO --url=tcp://localhost:61616 --persistent
*
* then it will try to call the following setters on the target object.
*
* target.setAckMode("AUTO");
* target.setURL(new URI("tcp://localhost:61616") );
* target.setPersistent(true);
*
* Notice the the proper conversion for the argument is determined by examining the
* setter arguement type.
*
* @param target the object that will have it's properties set
* @param args the commline options
* @return any arguments that are not valid options for the target
*/
public static String[] setOptions(Object target, String[] args) {
ArrayList<String> rc = new ArrayList<String>();
for (int i = 0; i < args.length; i++) {
if (args[i] == null) {
continue;
}
if (args[i].startsWith("--")) {
// --options without a specified value are considered boolean
// flags that are enabled.
String value = "true";
String name = args[i].substring(2);
// if --option=value case
int p = name.indexOf("=");
if (p > 0) {
value = name.substring(p + 1);
name = name.substring(0, p);
}
// name not set, then it's an unrecognized option
if (name.length() == 0) {
rc.add(args[i]);
continue;
}
String propName = convertOptionToPropertyName(name);
if (!IntrospectionSupport.setProperty(target, propName, value)) {
rc.add(args[i]);
continue;
}
}
}
String r[] = new String[rc.size()];
rc.toArray(r);
return r;
}
/**
* converts strings like: test-enabled to testEnabled
*
* @param name
* @return
*/
private static String convertOptionToPropertyName(String name) {
String rc = "";
// Look for '-' and strip and then convert the subsequent char to
// uppercase
int p = name.indexOf("-");
while (p > 0) {
// strip
rc += name.substring(0, p);
name = name.substring(p + 1);
// can I convert the next char to upper?
if (name.length() > 0) {
rc += name.substring(0, 1).toUpperCase();
name = name.substring(1);
}
p = name.indexOf("-");
}
return rc + name;
}
}
Raw
ConsumerTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.io.IOException;
import java.util.Arrays;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* A simple tool for consuming messages
*
* @version $Revision: 1.1.1.1 $
*/
public class ConsumerTool implements MessageListener, ExceptionListener {
private boolean running;
private Session session;
private Destination destination;
private MessageProducer replyProducer;
private boolean pauseBeforeShutdown;
private boolean verbose = true;
private int maxiumMessages;
private String subject = "TOOL.DEFAULT";
private boolean topic;
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private boolean transacted;
private boolean durable;
private String clientId;
private int ackMode = Session.AUTO_ACKNOWLEDGE;
private String consumerName = "James";
private long sleepTime;
private long receiveTimeOut;
public static void main(String[] args) {
ConsumerTool consumerTool = new ConsumerTool();
String[] unknown = CommandLineSupport.setOptions(consumerTool, args);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
consumerTool.run();
}
public void run() {
try {
running = true;
System.out.println("Connecting to URL: " + url);
System.out.println("Consuming " + (topic ? "topic" : "queue") + ": " + subject);
System.out.println("Using a " + (durable ? "durable" : "non-durable") + " subscription");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
Connection connection = connectionFactory.createConnection();
if (durable && clientId != null && clientId.length() > 0 && !"null".equals(clientId)) {
connection.setClientID(clientId);
}
connection.setExceptionListener(this);
connection.start();
session = connection.createSession(transacted, ackMode);
if (topic) {
destination = session.createTopic(subject);
} else {
destination = session.createQueue(subject);
}
replyProducer = session.createProducer(null);
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
MessageConsumer consumer = null;
if (durable && topic) {
consumer = session.createDurableSubscriber((Topic)destination, consumerName);
} else {
consumer = session.createConsumer(destination);
}
if (maxiumMessages > 0) {
consumeMessagesAndClose(connection, session, consumer);
} else {
if (receiveTimeOut == 0) {
consumer.setMessageListener(this);
} else {
consumeMessagesAndClose(connection, session, consumer, receiveTimeOut);
}
}
} catch (Exception e) {
System.out.println("Caught: " + e);
e.printStackTrace();
}
}
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage)message;
if (verbose) {
String msg = txtMsg.getText();
if (msg.length() > 50) {
msg = msg.substring(0, 50) + "...";
}
System.out.println("Received: " + msg);
}
} else {
if (verbose) {
System.out.println("Received: " + message);
}
}
if (message.getJMSReplyTo() != null) {
replyProducer.send(message.getJMSReplyTo(), session.createTextMessage("Reply: " + message.getJMSMessageID()));
}
if (transacted) {
session.commit();
} else if (ackMode == Session.CLIENT_ACKNOWLEDGE) {
message.acknowledge();
}
} catch (JMSException e) {
System.out.println("Caught: " + e);
e.printStackTrace();
} finally {
if (sleepTime > 0) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
}
}
}
}
public synchronized void onException(JMSException ex) {
System.out.println("JMS Exception occured. Shutting down client.");
running = false;
}
synchronized boolean isRunning() {
return running;
}
protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer) throws JMSException, IOException {
System.out.println("We are about to wait until we consume: " + maxiumMessages + " message(s) then we will shutdown");
for (int i = 0; i < maxiumMessages && isRunning();) {
Message message = consumer.receive(1000);
if (message != null) {
i++;
onMessage(message);
}
}
System.out.println("Closing connection");
consumer.close();
session.close();
connection.close();
if (pauseBeforeShutdown) {
System.out.println("Press return to shut down");
System.in.read();
}
}
protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer, long timeout) throws JMSException, IOException {
System.out.println("We will consume messages while they continue to be delivered within: " + timeout + " ms, and then we will shutdown");
Message message;
while ((message = consumer.receive(timeout)) != null) {
onMessage(message);
}
System.out.println("Closing connection");
consumer.close();
session.close();
connection.close();
if (pauseBeforeShutdown) {
System.out.println("Press return to shut down");
System.in.read();
}
}
public void setAckMode(String ackMode) {
if ("CLIENT_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.CLIENT_ACKNOWLEDGE;
}
if ("AUTO_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.AUTO_ACKNOWLEDGE;
}
if ("DUPS_OK_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.DUPS_OK_ACKNOWLEDGE;
}
if ("SESSION_TRANSACTED".equals(ackMode)) {
this.ackMode = Session.SESSION_TRANSACTED;
}
}
public void setClientId(String clientID) {
this.clientId = clientID;
}
public void setConsumerName(String consumerName) {
this.consumerName = consumerName;
}
public void setDurable(boolean durable) {
this.durable = durable;
}
public void setMaxiumMessages(int maxiumMessages) {
this.maxiumMessages = maxiumMessages;
}
public void setPauseBeforeShutdown(boolean pauseBeforeShutdown) {
this.pauseBeforeShutdown = pauseBeforeShutdown;
}
public void setPassword(String pwd) {
this.password = pwd;
}
public void setReceiveTimeOut(long receiveTimeOut) {
this.receiveTimeOut = receiveTimeOut;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTopic(boolean topic) {
this.topic = topic;
}
public void setQueue(boolean queue) {
this.topic = !queue;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public void setVerbose(boolean verbose) {
this.verbose = verbose;
}
}
Raw
ConsumerTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.io.IOException;
import java.util.Arrays;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* A simple tool for consuming messages
*
* @version $Revision: 1.1.1.1 $
*/
public class ConsumerTool implements MessageListener, ExceptionListener {
private Log log = LogFactory.getLog(getClass());
private boolean running;
private Session session;
private Destination destination;
private MessageProducer replyProducer;
private boolean pauseBeforeShutdown;
private boolean verbose = true;
private int maxiumMessages;
private String subject = "TOOL.DEFAULT";
private boolean topic;
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private boolean transacted;
private boolean durable;
private String clientId;
private int ackMode = Session.AUTO_ACKNOWLEDGE;
private String consumerName = "James";
private long sleepTime;
private long receiveTimeOut;
public static void main(String[] args) {
ConsumerTool consumerTool = new ConsumerTool();
String[] unknown = CommandLineSupport.setOptions(consumerTool, args);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
consumerTool.run();
}
public void run() {
try {
running = true;
log.info("Connecting to URL: " + url);
log.info("Consuming " + (topic ? "topic" : "queue") + ": " + subject);
log.info("Using a " + (durable ? "durable" : "non-durable") + " subscription");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
Connection connection = connectionFactory.createConnection();
if (durable && clientId != null && clientId.length() > 0 && !"null".equals(clientId)) {
connection.setClientID(clientId);
}
connection.setExceptionListener(this);
connection.start();
session = connection.createSession(transacted, ackMode);
if (topic) {
destination = session.createTopic(subject);
} else {
destination = session.createQueue(subject);
}
replyProducer = session.createProducer(null);
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
MessageConsumer consumer = null;
if (durable && topic) {
consumer = session.createDurableSubscriber((Topic)destination, consumerName);
} else {
consumer = session.createConsumer(destination);
}
if (maxiumMessages > 0) {
consumeMessagesAndClose(connection, session, consumer);
} else {
if (receiveTimeOut == 0) {
consumer.setMessageListener(this);
} else {
consumeMessagesAndClose(connection, session, consumer, receiveTimeOut);
}
}
} catch (Exception e) {
log.info("Caught: " + e);
e.printStackTrace();
}
}
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage)message;
if (verbose) {
String msg = txtMsg.getText();
if (msg.length() > 50) {
msg = msg.substring(0, 50) + "...";
}
log.info("Received: " + msg);
}
} else {
if (verbose) {
log.info("Received: " + message);
}
}
if (message.getJMSReplyTo() != null) {
replyProducer.send(message.getJMSReplyTo(), session.createTextMessage("Reply: " + message.getJMSMessageID()));
}
if (transacted) {
session.commit();
} else if (ackMode == Session.CLIENT_ACKNOWLEDGE) {
message.acknowledge();
}
} catch (JMSException e) {
log.info("Caught: " + e);
e.printStackTrace();
} finally {
if (sleepTime > 0) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
}
}
}
}
public synchronized void onException(JMSException ex) {
log.info("JMS Exception occured. Shutting down client.");
running = false;
}
synchronized boolean isRunning() {
return running;
}
protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer) throws JMSException, IOException {
log.info("We are about to wait until we consume: " + maxiumMessages + " message(s) then we will shutdown");
for (int i = 0; i < maxiumMessages && isRunning();) {
Message message = consumer.receive(1000);
if (message != null) {
i++;
onMessage(message);
}
}
log.info("Closing connection");
consumer.close();
session.close();
connection.close();
if (pauseBeforeShutdown) {
log.info("Press return to shut down");
System.in.read();
}
}
protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer, long timeout) throws JMSException, IOException {
log.info("We will consume messages while they continue to be delivered within: " + timeout + " ms, and then we will shutdown");
Message message;
while ((message = consumer.receive(timeout)) != null) {
onMessage(message);
}
log.info("Closing connection");
consumer.close();
session.close();
connection.close();
if (pauseBeforeShutdown) {
log.info("Press return to shut down");
System.in.read();
}
}
public void setAckMode(String ackMode) {
if ("CLIENT_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.CLIENT_ACKNOWLEDGE;
}
if ("AUTO_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.AUTO_ACKNOWLEDGE;
}
if ("DUPS_OK_ACKNOWLEDGE".equals(ackMode)) {
this.ackMode = Session.DUPS_OK_ACKNOWLEDGE;
}
if ("SESSION_TRANSACTED".equals(ackMode)) {
this.ackMode = Session.SESSION_TRANSACTED;
}
}
public void setClientId(String clientID) {
this.clientId = clientID;
}
public void setConsumerName(String consumerName) {
this.consumerName = consumerName;
}
public void setDurable(boolean durable) {
this.durable = durable;
}
public void setMaxiumMessages(int maxiumMessages) {
this.maxiumMessages = maxiumMessages;
}
public void setPauseBeforeShutdown(boolean pauseBeforeShutdown) {
this.pauseBeforeShutdown = pauseBeforeShutdown;
}
public void setPassword(String pwd) {
this.password = pwd;
}
public void setReceiveTimeOut(long receiveTimeOut) {
this.receiveTimeOut = receiveTimeOut;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTopic(boolean topic) {
this.topic = topic;
}
public void setQueue(boolean queue) {
this.topic = !queue;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public void setVerbose(boolean verbose) {
this.verbose = verbose;
}
}
Raw
corp-activemq.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="BROKERNAME">
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#mssql-jtds-ds" adapter="#sqlserver-adapter"/>
</persistenceAdapter>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<tempUsage>
<tempUsage limit="50 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<transportConnectors>
<transportConnector name="openwire" uri="tcp://BROKERHOST:61616"/>
</transportConnectors>
<plugins>
<jaasAuthenticationPlugin configuration="activemq-domain" />
</plugins>
</broker>
<!-- JTDS (DBCP) DataSource Sample Setup -->
<bean id="mssql-jtds-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="net.sourceforge.jtds.jdbcx.JtdsDataSource"/>
<property name="url" value="jdbc:jtds:sqlserver://HOST:PORT;DatabaseName=ACTIVEMQ"/>
<property name="username" value="blabl"/>
<property name="password" value="blabla"/>
<property name="poolPreparedStatements" value="false"/>
<property name="maxWait" value="20"/>
<property name="maxIdle" value="10"/>
<property name="maxActive" value="20"/>
<property name="validationQuery" value="select cast('10' as integer)"/>
<property name="testOnBorrow" value="false"/>
<property name="defaultAutoCommit" value="false"/>
</bean>
<bean id="sqlserver-adapter" class="ch.bedag.ste.opensource.activemq.jdbc.adapter.SqlServerJDBCAdapter"/>
<bean id="ds-exporter" class="org.springframework.jmx.export.MBeanExporter" lazy-init="false">
<property name="beans">
<map>
<entry key="DataSource:name=mssql" value-ref="mssql-jtds-ds"/>
</map>
</property>
<property name="registrationBehaviorName" value="REGISTRATION_REPLACE_EXISTING"/>
<property name="assembler">
<bean class="org.springframework.jmx.export.assembler.MethodNameBasedMBeanInfoAssembler">
<property name="managedMethods">
<value>getDriverClassName,getDefaultAutoCommit,getLogAbandoned,setLogAbandoned,getMaxActive,getMaxIdle,getMaxOpenPreparedStatements,getMaxWait,getMinIdle,getNumActive,getNumIdle,isPoolPreparedStatements,getUrl,getUsername,getValidationQuery,getTestOnBorrow,getTestOnReturn,getTestWhileIdle</value>
</property>
</bean>
</property>
</bean>
</beans>
Raw
create-broker-url
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
URL=for line in $( ec2-with-group ); do
echo "tcp://$line:61616"
done | tr "\n" ","
URL="static://$URL"
echo $URL > $BROKER_URL_FILE
Raw
ec2-sel-by-group
#!/bin/bash
ec2din --xml | \
xmlstarlet sel -t \
-m "//reservationSet/item" \
-i "groupSet/item/groupId[string()='$GROUP_NAME']" \
-m "instancesSet/item" \
-i "instanceState/name[string()='running']" \
-v "instanceId" -n | \
sed \ # -> http://www.grymoire.com/Unix/Sed.html#toc-uh-30
-e 's/#.*//' \ # remove every character from the "#" to the end of the line
-e 's/[ ^I]*$//' \ # delete all blank lines
-e '/^$/ d' # remove all blanks and tabs immediately before the end of line
Raw
ec2-start-hyperic
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
ec2-run-instances -k manual -t m1.small -z AvailabilityZone -g hq-server -g default ami-377a9d5e
-dINITPWORD=bedag
Raw
ec2din.xml
<?xml version="1.0"?>
<DescribeInstancesResponse>
<requestId>29c10b8c-ead4-47c5-978d-ebe19cfc270e</requestId>
<reservationSet>
<item>
<reservationId>r-304bdb59</reservationId>
<ownerId>001523653591</ownerId>
<groupSet>
<item>
<groupId>activemq-broker</groupId>
</item>
<item>
<groupId>default</groupId>
</item>
</groupSet>
<instancesSet>
<item>
<instanceId>i-1fd9a976</instanceId>
<imageId>ami-9209eefb</imageId>
<instanceState>
<code>16</code>
<name>running</name>
</instanceState>
<privateDnsName>domU-12-31-39-00-B4-B3.compute-1.internal</privateDnsName>
<dnsName>ec2-67-202-51-38.compute-1.amazonaws.com</dnsName>
<reason/>
<keyName>manual</keyName>
<amiLaunchIndex>0</amiLaunchIndex>
<productCodes/>
<instanceType>m1.small</instanceType>
<launchTime>2009-05-11T07:52:28.000Z</launchTime>
<placement>
<availabilityZone>us-east-1b</availabilityZone>
</placement>
<kernelId>aki-a71cf9ce</kernelId>
<ramdiskId>ari-a51cf9cc</ramdiskId>
</item>
</instancesSet>
</item>
<item>
<reservationId>check</reservationId>
<ownerId>001523653591</ownerId>
<groupSet>
<item>
<groupId>activemq-broker</groupId>
</item>
<item>
<groupId>default</groupId>
</item>
</groupSet>
<instancesSet>
<item>
<instanceId>i-check</instanceId>
<imageId>ami-9209eefb</imageId>
<instanceState>
<code>16</code>
<name>something</name>
</instanceState>
<privateDnsName>domU-12-31-39-00-B4-B3.compute-1.internal</privateDnsName>
<dnsName>ec2-67-202-51-38.compute-1.amazonaws.com</dnsName>
<reason/>
<keyName>manual</keyName>
<amiLaunchIndex>0</amiLaunchIndex>
<productCodes/>
<instanceType>m1.small</instanceType>
<launchTime>2009-05-11T07:52:28.000Z</launchTime>
<placement>
<availabilityZone>us-east-1b</availabilityZone>
</placement>
<kernelId>aki-a71cf9ce</kernelId>
<ramdiskId>ari-a51cf9cc</ramdiskId>
</item>
</instancesSet>
</item>
</reservationSet>
</DescribeInstancesResponse>
Raw
EmbeddedBroker.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import org.apache.activemq.broker.BrokerService;
/**
* This example demonstrates how to run an embedded broker inside your Java code
*
* @version $Revision: 565003 $
*/
public final class EmbeddedBroker {
private EmbeddedBroker() {
}
public static void main(String[] args) throws Exception {
BrokerService broker = new BrokerService();
broker.setUseJmx(true);
broker.addConnector("tcp://localhost:61616");
broker.start();
// now lets wait forever to avoid the JVM terminating immediately
Object lock = new Object();
synchronized (lock) {
lock.wait();
}
}
}
Raw
find-shebangs
#!/bin/bash
#
# look for all the files in a directory, and examine the first line
# of each for a shebang ("#!")
#
# Aspects
# - recursiv
# - stdout from head is piped into grep, and grep is silent (-q). (only the status from grep is used)
# - the filename is echoed for your "convenience" ;) (ie: not necessary!)
# - check for files owned by $USER
# -
#
#
#
#
# http://www.linuxquestions.org/questions/linux-general-1/bash-script-to-find-files-with-shebang-and-then-chmod-686031/
#
set -o errexit # Exit when simple command fails.
set -o nounset # Trigger error when expanding unset variables.
find dir/ -user $UID -type f | while read fn; do head -n1 "$fn" | grep -q "^#\!" && echo "$fn" && chmod 755 "$fn"; done
Raw
inmemory-activemq
#!/bin/bash
set -o errexit # Exit when simple command fails. Same as `set -e'.
set -o nounset # Trigger error when expanding unset variables.
source btheaders # Materialized definitions (variables) of paths on filesystem
case "$1" in
start)
ACTIVEMQ_CONFIG=$BTCONFIG_HOME/inmemory-activemq.xml
STD_LOG=$BTLOGS_HOME/inmemory-activemq.log
echo "$ACTIVEMQ_CONFIG"
echo "$STD_LOG"
time $HOME/apache-activemq-5.2.0/bin/activemq xbean:file:$ACTIVEMQ_CONFIG > $STD_LOG &
;;
stop)
kill `ps -ef | grep apache-activemq-5\.2\.0 | grep -v grep | grep -v kill | awk '{print $2}'`
;;
status)
status anacron
;;
restart)
$0 start
$0 stop
;;
*)
echo $"Usage: $0 {start|stop|restart|status}"
exit 1
esac
Raw
inmemory-activemq.xml
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<!-- Allows us to use system properties as variables in this configuration file -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:///${activemq.base}/conf/credentials.properties</value>
</property>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-a" dataDirectory="${activemq.base}/data">
<!-- Use the following to configure how ActiveMQ is exposed in JMX -->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers
<networkConnector name="default-nc" uri="multicast://default"/>
-->
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
</networkConnectors>
<persistenceAdapter>
<amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/data" maxFileLength="20 mb"/>
</persistenceAdapter>
<!-- The maximum about of space the broker will use before slowing down producers -->
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage limit="200 mb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="1 gb" name="foo"/>
</storeUsage>
<tempUsage>
<tempUsage limit="500 mb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<!-- The transport connectors ActiveMQ will listen to -->
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61616" discoveryUri="multicast://default"/>
</transportConnectors>
</broker>
<!-- An embedded servlet engine for serving up the Admin console -->
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8161"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>
<!-- Embedded Derby DataSource Sample Setup -->
<!--
<bean id="derby-ds" class="org.apache.derby.jdbc.EmbeddedDataSource">
<property name="databaseName" value="derbydb"/>
<property name="createDatabase" value="create"/>
</bean>
-->
</beans>
Raw
install.sh
#!/bin/bash
#set -o errexit # Exit when simple command fails. Same as `set -e'.
#set -o nounset # Trigger error when expanding unset variables.
apt-get -y install ant ant-optional screen python
mkdir -p ~/downloads
cd ~/downloads
[ ! -e ~/downloads/bootstrap.py ] && { wget http://svn.zope.org/*checkout*/zc.buildout/trunk/bootstrap/bootstrap.py; }
[ ! -e ~/downloads/jtds-1.2.2.jar ] && { wget https://m2proxy.atlassian.com/repository/public/net/sourceforge/jtds/jtds/1.2.2/jtds-1.2.2.jar; }
[ ! -e ~/downloads/hyperic-hq-installer-4.1.1-1045-x86-linux.tgz ] && { wget http://internap.dl.sourceforge.net/sourceforge/hyperic-hq/hyperic-hq-installer-4.1.1-1045-x86-linux.tgz; }
adduser --system hyperic
/home/hyperic/agent-4.1.1/bin/hq-agent.sh restart
Raw
instances.py
#!/usr/bin/env python
# http://code.google.com/p/boto/source/browse/trunk/boto/ec2/connection.py
from __future__ import with_statement
import os
import subprocess
import optparse
import yaml
import logging
from boto.ec2.connection import EC2Connection
log = logging.getLogger('activemq_stress')
roles = yaml.load("""
broker:
ami: ami-9209eefb
groups: [activemq-broker]
instance_type : m1.small
participant:
ami: ami-9209eefb
groups: []
instance_type : m1.small
hyperic:
ami: ami-377a9d5e
groups: [hq-server]
instance_type : m1.small
sqlserver:
ami: ami-c64daaaf
groups: [sql-server]
instance_type : m1.large
""")
def main():
usage = "usage: %prog [options]"
parser = optparse.OptionParser(usage)
parser.add_option("--list", dest="list",
action="store_true",
default=True,
help="""
Lists all running instances, how long they have been running, and their type
""")
parser.add_option("-l", "--launch", dest="launch",
action="store",
help="Start an image [broker|participant|sqlserver|hyperic]")
parser.add_option("-t", "--terminate", dest="terminate",
action="store",
help="Shutdown an instance (id) or a member of a role [broker|participant|sqlserver|hyperic]")
# TODO how to add default args?
options, args = parser.parse_args()
# find out about list comprehensions to make
# launch and shutdown nice and short.
log.debug('options: %s' % options)
if options.launch:
launch(options.launch)
if options.terminate:
terminate(options.terminate)
if options.list:
list_instances()
def launch(rolename):
if rolename not in roles:
raise Exception, "failed to find a suitable role: %s" % rolename
role = roles[rolename]
conn = aws_connect()
conn.run_instances(role['ami'],
key_name='manual',
security_groups=role['groups'],
instance_type=role['instance_type'])
def terminate(rolename_or_instance):
print "terminating: %s" % rolename_or_instance
conn = aws_connect()
if rolename_or_instance in roles: # is it a rolename?
role = roles[rolename_or_instance]['ami']
log.debug("terminating all rolemembers for: %s" % role)
instances = []
for reservation in conn.get_all_instances():
for instance in reservation.instances:
if instance.security_group in role['groups']:
instances += instance.id
if instances:
conn.terminate_instances(instances)
else: # it might be an instance name
result = conn.get_all_instances(rolename_or_instance)
if result.isError:
raise Exception, "failed to find a role or instance with id: %s" % rolename_or_instance
else:
conn.terminate_instances(rolename_or_instance) # it's an instance_id
def list_instances():
"""Lists all running instances (id), how long they have been running, and their type"""
conn = aws_connect()
reservations = conn.get_all_instances()
print yaml.safe_dump(transform_reservations(reservations))
def transform_reservations(reservations, all=False):
if all:
return dump_as_dict(reservations)
instances_by_id = {}
for rsv in reservations:
groups = [group.id for group in rsv.groups]
for instance in rsv.instances:
if instance.state is 'terminated':
pass
# construct something of the form (yaml format)
# <instance_id>:
# - groups: [ <group_id>, ... ]
# - ami: <ami_id>
# - state: <run_state>
instance_hash = {}
instance_hash['image_id'] = instance.image_id
instance_hash['state'] = instance.state
instance_hash['groups'] = groups
instance_hash['launch_time'] = instance.launch_time
instances_by_id[instance.id] = instance_hash
return instances_by_id
def aws_connect():
awssecret = os.path.join(os.getenv("HOME"), '.awssecret')
if os.path.exists(awssecret) != True:
raise Exception, "path to %s does not exist" % awssecret # what kind of exception
with open(awssecret, 'r') as f:
lines = f.readlines()
access_key = lines[0].strip()
access_secret = lines[1].strip()
log.debug("%s -- %s" % (access_key, access_secret))
conn = EC2Connection(access_key, access_secret)
return conn
# inspired from http://svn.pyyaml.org/pyyaml-legacy/trunk/experimental/XmlYaml/convertyaml_map.py
filtered_attrs = ['connection',
'is_truncated',
'marker',
'markers',
'next_token',
'reservationSet',
'requestId',
'status',
'DescribeInstancesResponse',
'groupSet',
'item',
'instancesSet' ]
def dump_as_dict(obj):
fqn = (lambda c: c.__module__ + "." + c.__name__)(obj.__class__)
log.debug("processing: %s" % fqn)
if obj.__class__.__module__.find("boto") != 0:
return obj
# copy __dict__ (filter out unnecessary stuff)
objDict = dict([ (key,value) for key,value in obj.__dict__.items() if key not in filtered_attrs ])
for key,value in obj.__dict__.items():
if key not in filtered_attrs:
objDict[key] = dump_as_dict(value)
# is it a list?
if isinstance(obj, list):
if objDict:
objDict['values'] = [dump_as_dict(e) for e in obj]
return objDict
else:
return [dump_as_dict(e) for e in obj]
log.debug("leaving: %s with %s" % (fqn, objDict))
return objDict
if __name__ == "__main__":
main()
Raw
INSTRUCTIONS
How to actually run the things here
# get the latest and greatest on your machine
apt-get update; apt-get -y upgrade; apt-get -y install git-core
# get the actual code to run on the machine
git clone git://gist.github.com/85430.git node-config; node-config/install.sh
# install.sh goes haywire and downloads some things (if necessary),
# and in general prepares the system.
now we can use the activemq command to start/stop the servers
activemq
-u --up Start the broker
-d --down Stop the broker
-nodb --no-database run the broker only on the filesystem (note that it's basically possible to run two
brokers on the same machine, but the -c flag will not take it into account.)
-st --status show the running brokers on all
-c --connection-url This will display the connection url necessary to connect to the brokers.
will output a correct string for all machines running with a security group named activemq-broker
the answer on the shell is going to be the name of this broker and the pid
No for a real run, we also have to start the hyperic server and the database server
as soon as these guys have been started, we can turn our attention to the stress-test script
stress-test
The stress test invokes the appropriate java commands to utilize the java library for testing
Raw
org.eclipse.jdt.core.prefs
#Wed May 13 20:55:37 ICT 2009
eclipse.preferences.version=1
org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5
org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
org.eclipse.jdt.core.compiler.compliance=1.5
org.eclipse.jdt.core.compiler.debug.lineNumber=generate
org.eclipse.jdt.core.compiler.debug.localVariable=generate
org.eclipse.jdt.core.compiler.debug.sourceFile=generate
org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
org.eclipse.jdt.core.compiler.source=1.5
Raw
org.maven.ide.eclipse.prefs
#Wed May 13 16:29:26 ICT 2009
activeProfiles=
eclipse.preferences.version=1
fullBuildGoals=process-test-resources
includeModules=false
resolveWorkspaceProjects=true
resourceFilterGoals=process-resources resources\:testResources
skipCompilerPlugin=true
version=1
Raw
paste-bin.txt
apt-get update; apt-get -y upgrade; apt-get -y install git-core ant ant-optional screen
apt-get install postgres
git clone git://gist.github.com/85430.git node-config; node-config/install.sh
wget https://m2proxy.atlassian.com/repository/public/net/sourceforge/jtds/jtds/1.2.2/jtds-1.2.2.jar
:8161/admin/queues.jsp
:7080/
INITPWORD=
Raw
path.sh
#!/bin/bash
#
# Converting a Relative Path to an Absolute Path
# with support for non-existent paths.
#
# Credits: http://www.robertpeaslee.com/index.php/converting-a-relative-path-to-an-absolute-path-in-bash/
#
if [ $# -eq 0 ]; then
echo "Usage: $0 "
exit 1
fi
if [ -d $1 ]; then
# Paramater is an existing directory. Print it using the method in the script above.
echo "Absolute path: `cd $1; pwd`"
elif [[ -e $1 && ! -d $1 ]]; then
# File already exists and isn't a directory. Be more safe with the conversion.
mkdir $1$$ 2> /dev/null
if[ $? -ne 0 ]; then
echo "We cannot conver this path without write permissions to the path's destination."
exit 1
fi
# I don't want to escape the quotes. This is already ugly, anyway.
dirName=`cd $1$$; pwd | awk -F"$$" {'print $1'}`
echo "Absolute path: $dirName"
rm -r $1$$
else
# File doesn't exist, begin unelegant conversion
mkdir $1 2> /dev/null
if [ $? -ne 0 ]; then
echo "We cannot convert this path without write permissions to the path's destination."
exit 1
fi
echo "Absolute path: `cd $1; pwd`"
rm -r $1
fi
Raw
pfind
#!/bin/bash
set -o errexit # Exit when simple command fails.
set -o nounset # Trigger error when expanding unset variables.
if [ $# -eq 0 ]; then
program=`basename "$BASH_SOURCE"`
echo " $program <grep-name> [ps-opts]"
else
regex=$1
shift
ps $* | grep $regex | grep -v grep | grep -v $0
fi
#ps aux | grep $1 | grep -v grep | grep -v $0
Raw
pkill
#!/bin/bash
if [ $# -eq 0 ]; then
program=`basename $0`
echo " $program <grep-name>"
else
regex=$1
shift
kill $@ `ps aux | grep $regex | grep -v grep | grep -v $0 | awk '{print $2}'`
fi
#ps aux | grep $1 | grep -v grep | grep -v $0
Raw
pom.xml
<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.digitalstrider.tools</groupId>
<artifactId>jms-utils</artifactId>
<name>JMS Utils</name>
<version>0.0.1-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.2.0</version>
</dependency>
</dependencies>
</project>
Raw
ProducerAndConsumerTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import java.util.HashSet;
import javax.jms.MessageListener;
/**
* A simple tool for producing and consuming messages
*
* @version $Revision: 1.1.1.1 $
*/
public class ProducerAndConsumerTool extends ConsumerTool implements MessageListener {
public static void main(String[] args) {
ConsumerTool consumerTool = new ConsumerTool();
String[] unknown = CommandLineSupport.setOptions(consumerTool, args);
HashSet<String> set1 = new HashSet<String>(Arrays.asList(unknown));
ProducerTool producerTool = new ProducerTool();
unknown = CommandLineSupport.setOptions(producerTool, args);
HashSet<String> set2 = new HashSet<String>(Arrays.asList(unknown));
set1.retainAll(set2);
if (set1.size() > 0) {
System.out.println("Unknown options: " + set1);
System.exit(-1);
}
consumerTool.run();
producerTool.run();
}
}
Raw
ProducerTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import java.util.Date;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.util.IndentPrinter;
/**
* A simple tool for publishing messages
*
* @version $Revision: 1.2 $
*/
public class ProducerTool {
private Destination destination;
private int messageCount = 10;
private long sleepTime;
private boolean verbose = true;
private int messageSize = 255;
private long timeToLive;
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private String subject = "TOOL.DEFAULT";
private boolean topic;
private boolean transacted;
private boolean persistent;
public static void main(String[] args) {
ProducerTool producerTool = new ProducerTool();
String[] unknown = CommandLineSupport.setOptions(producerTool, args);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
producerTool.run();
}
public void run() {
Connection connection = null;
try {
System.out.println("Connecting to URL: " + url);
System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + subject);
System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
System.out.println("Sleeping between publish " + sleepTime + " ms");
if (timeToLive != 0) {
System.out.println("Messages time to live " + timeToLive + " ms");
}
// Create the connection.
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
connection = connectionFactory.createConnection();
connection.start();
// Create the session
Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
if (topic) {
destination = session.createTopic(subject);
} else {
destination = session.createQueue(subject);
}
// Create the producer.
MessageProducer producer = session.createProducer(destination);
if (persistent) {
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
} else {
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}
if (timeToLive != 0) {
producer.setTimeToLive(timeToLive);
}
// Start sending messages
sendLoop(session, producer);
System.out.println("Done.");
// Use the ActiveMQConnection interface to dump the connection
// stats.
ActiveMQConnection c = (ActiveMQConnection)connection;
c.getConnectionStats().dump(new IndentPrinter());
} catch (Exception e) {
System.out.println("Caught: " + e);
e.printStackTrace();
} finally {
try {
connection.close();
} catch (Throwable ignore) {
}
}
}
protected void sendLoop(Session session, MessageProducer producer) throws Exception {
for (int i = 0; i < messageCount || messageCount == 0; i++) {
TextMessage message = session.createTextMessage(createMessageText(i));
if (verbose) {
String msg = message.getText();
if (msg.length() > 50) {
msg = msg.substring(0, 50) + "...";
}
System.out.println("Sending message: " + msg);
}
producer.send(message);
if (transacted) {
session.commit();
}
Thread.sleep(sleepTime);
}
}
private String createMessageText(int index) {
StringBuffer buffer = new StringBuffer(messageSize);
buffer.append("Message: " + index + " sent at: " + new Date());
if (buffer.length() > messageSize) {
return buffer.substring(0, messageSize);
}
for (int i = buffer.length(); i < messageSize; i++) {
buffer.append(' ');
}
return buffer.toString();
}
public void setPersistent(boolean durable) {
this.persistent = durable;
}
public void setMessageCount(int messageCount) {
this.messageCount = messageCount;
}
public void setMessageSize(int messageSize) {
this.messageSize = messageSize;
}
public void setPassword(String pwd) {
this.password = pwd;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
public void setTopic(boolean topic) {
this.topic = topic;
}
public void setQueue(boolean queue) {
this.topic = !queue;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public void setVerbose(boolean verbose) {
this.verbose = verbose;
}
}
Raw
ProducerTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import java.util.Date;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.util.IndentPrinter;
/**
* A simple tool for publishing messages
*
* @version $Revision: 1.2 $
*/
public class ProducerTool {
private Destination destination;
private int messageCount = 10;
private long sleepTime;
private boolean verbose = true;
private int messageSize = 255;
private long timeToLive;
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private String subject = "TOOL.DEFAULT";
private boolean topic;
private boolean transacted;
private boolean persistent;
public static void main(String[] args) {
ProducerTool producerTool = new ProducerTool();
String[] unknown = CommandLineSupport.setOptions(producerTool, args);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
producerTool.run();
}
public void run() {
Connection connection = null;
try {
System.out.println("Connecting to URL: " + url);
System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + subject);
System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
System.out.println("Sleeping between publish " + sleepTime + " ms");
if (timeToLive != 0) {
System.out.println("Messages time to live " + timeToLive + " ms");
}
// Create the connection.
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
connection = connectionFactory.createConnection();
connection.start();
// Create the session
Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
if (topic) {
destination = session.createTopic(subject);
} else {
destination = session.createQueue(subject);
}
// Create the producer.
MessageProducer producer = session.createProducer(destination);
if (persistent) {
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
} else {
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}
if (timeToLive != 0) {
producer.setTimeToLive(timeToLive);
}
// Start sending messages
sendLoop(session, producer);
System.out.println("Done.");
// Use the ActiveMQConnection interface to dump the connection
// stats.
ActiveMQConnection c = (ActiveMQConnection)connection;
c.getConnectionStats().dump(new IndentPrinter());
} catch (Exception e) {
System.out.println("Caught: " + e);
e.printStackTrace();
} finally {
try {
connection.close();
} catch (Throwable ignore) {
}
}
}
protected void sendLoop(Session session, MessageProducer producer) throws Exception {
for (int i = 0; i < messageCount || messageCount == 0; i++) {
TextMessage message = session.createTextMessage(createMessageText(i));
if (verbose) {
String msg = message.getText();
if (msg.length() > 50) {
msg = msg.substring(0, 50) + "...";
}
System.out.println("Sending message: " + msg);
}
producer.send(message);
if (transacted) {
session.commit();
}
Thread.sleep(sleepTime);
}
}
private String createMessageText(int index) {
StringBuffer buffer = new StringBuffer(messageSize);
buffer.append("Message: " + index + " sent at: " + new Date());
if (buffer.length() > messageSize) {
return buffer.substring(0, messageSize);
}
for (int i = buffer.length(); i < messageSize; i++) {
buffer.append(' ');
}
return buffer.toString();
}
public void setPersistent(boolean durable) {
this.persistent = durable;
}
public void setMessageCount(int messageCount) {
this.messageCount = messageCount;
}
public void setMessageSize(int messageSize) {
this.messageSize = messageSize;
}
public void setPassword(String pwd) {
this.password = pwd;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
public void setTopic(boolean topic) {
this.topic = topic;
}
public void setQueue(boolean queue) {
this.topic = !queue;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public void setVerbose(boolean verbose) {
this.verbose = verbose;
}
}
Raw
profile.sh
bind '"\M-d": backward-kill-word'
export PATH=$PATH:`pwd`
pwd=`which pwd`
alias ls='ls --color'
function ll () {
ls -Gl $1
$pwd
}
function la () {
ls -Gla $1
$pwd
}
alias l=ll
Raw
RequesterTool.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import java.util.Date;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.util.IndentPrinter;
/**
* A simple tool for publishing messages
*
* @version $Revision: 1.2 $
*/
public class RequesterTool {
private int messageCount = 10;
private long sleepTime;
private boolean verbose = true;
private int messageSize = 255;
private long timeToLive;
private String subject = "TOOL.DEFAULT";
private String replySubject;
private boolean topic;
private String user = ActiveMQConnection.DEFAULT_USER;
private String password = ActiveMQConnection.DEFAULT_PASSWORD;
private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private boolean transacted;
private boolean persistent;
private String clientId;
private Destination destination;
private Destination replyDest;
private MessageProducer producer;
private MessageConsumer consumer;
private Session session;
public static void main(String[] args) {
RequesterTool requesterTool = new RequesterTool();
String[] unknown = CommandLineSupport.setOptions(requesterTool, args);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
requesterTool.run();
}
public void run() {
Connection connection = null;
try {
System.out.println("Connecting to URL: " + url);
System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + subject);
System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
System.out.println("Sleeping between publish " + sleepTime + " ms");
// Create the connection
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
connection = connectionFactory.createConnection();
if (persistent && clientId != null && clientId.length() > 0 && !"null".equals(clientId)) {
connection.setClientID(clientId);
}
connection.start();
// Create the Session
session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
// And the Destinations..
if (topic) {
destination = session.createTopic(subject);
if (replySubject == null || replySubject.equals("")) {
replyDest = session.createTemporaryTopic();
} else {
replyDest = session.createTopic(replySubject);
}
} else {
destination = session.createQueue(subject);
if (replySubject == null || replySubject.equals("")) {
replyDest = session.createTemporaryQueue();
} else {
replyDest = session.createQueue(replySubject);
}
}
System.out.println("Reply Destination: " + replyDest);
// Create the producer
producer = session.createProducer(destination);
if (persistent) {
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
} else {
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}
if (timeToLive != 0) {
System.out.println("Messages time to live " + timeToLive + " ms");
producer.setTimeToLive(timeToLive);
}
// Create the reply consumer
consumer = session.createConsumer(replyDest);
// Start sending reqests.
requestLoop();
System.out.println("Done.");
// Use the ActiveMQConnection interface to dump the connection
// stats.
ActiveMQConnection c = (ActiveMQConnection)connection;
c.getConnectionStats().dump(new IndentPrinter());
} catch (Exception e) {
System.out.println("Caught: " + e);
e.printStackTrace();
} finally {
try {
connection.close();
} catch (Throwable ignore) {
}
}
}
protected void requestLoop() throws Exception {
for (int i = 0; i < messageCount || messageCount == 0; i++) {
TextMessage message = session.createTextMessage(createMessageText(i));
message.setJMSReplyTo(replyDest);
if (verbose) {
String msg = message.getText();
if (msg.length() > 50) {
msg = msg.substring(0, 50) + "...";
}
System.out.println("Sending message: " + msg);
}
producer.send(message);
if (transacted) {
session.commit();
}
System.out.println("Waiting for reponse message...");
Message message2 = consumer.receive();
if (message2 instanceof TextMessage) {
System.out.println("Reponse message: " + ((TextMessage)message2).getText());
} else {
System.out.println("Reponse message: " + message2);
}
if (transacted) {
session.commit();
}
Thread.sleep(sleepTime);
}
}
/**
* @param i
* @return
*/
private String createMessageText(int index) {
StringBuffer buffer = new StringBuffer(messageSize);
buffer.append("Message: " + index + " sent at: " + new Date());
if (buffer.length() > messageSize) {
return buffer.substring(0, messageSize);
}
for (int i = buffer.length(); i < messageSize; i++) {
buffer.append(' ');
}
return buffer.toString();
}
public void setClientId(String clientId) {
this.clientId = clientId;
}
public void setPersistent(boolean durable) {
this.persistent = durable;
}
public void setMessageCount(int messageCount) {
this.messageCount = messageCount;
}
public void setMessageSize(int messageSize) {
this.messageSize = messageSize;
}
public void setPassword(String password) {
this.password = password;
}
public void setSleepTime(long sleepTime) {
this.sleepTime = sleepTime;
}
public void setSubject(String subject) {
this.subject = subject;
}
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
public void setTopic(boolean topic) {
this.topic = topic;
}
public void setQueue(boolean queue) {
this.topic = !queue;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public void setUrl(String url) {
this.url = url;
}
public void setUser(String user) {
this.user = user;
}
public void setVerbose(boolean verbose) {
this.verbose = verbose;
}
public void setReplySubject(String replySubject) {
this.replySubject = replySubject;
}
}
Raw
setup.py
#!/usr/bin/env python
from setuptools import setup
setup(name='activemq_stress',
version='0.0.1',
description='Stress testing tools for ActiveMQ >5.2.0',
long_description="""
""",
author='Daniel',
author_email='',
url='http://todo/',
packages=['activemq_stress'],
package_dir={'': 'src'},
install_requires=[
'boto==1.7a',
'PyYAML==3.05',
# 'Gnosis_Utils==1.2.1-a',
],
entry_points=("""
[console_scripts]
instances=activemq_stress.instances:main
activemq=activemq_stress.activemq:main
tryout=activemq_stress.tryout:main
"""),
classifiers=[
'Development Status :: 4 - Beta',
'License :: OSI Approved :: Python Software Foundation License',
'Environment :: Console',
'Environment :: Web Environment',
'Operating System :: MacOS :: MacOS X',
'Operating System :: Microsoft :: Windows',
'Operating System :: POSIX',
'Programming Language :: Python',
'Topic :: Software Development :: Commandline',
'Intended Audience :: Developers',
'Intended Audience :: System Administrators'
],
)
Raw
setup.py
#!/usr/bin/env python
from setuptools import setup
setup(name='com.digitalstrider.tools.syspack',
version='0.0.1',
description='Assorted scripts to make admin life easier',
long_description="""
""",
author='Daniel',
author_email='',
url='http://todo/',
packages=['src.powerpack'],
install_requires=[
'boto==1.7a',
'setuptools-git',
],
classifiers=[
'Development Status :: 4 - Beta',
'License :: OSI Approved :: Python Software Foundation License',
'Environment :: Console',
'Environment :: Web Environment',
'Operating System :: MacOS :: MacOS X',
'Operating System :: Microsoft :: Windows',
'Operating System :: POSIX',
'Programming Language :: Python',
'Topic :: Software Development :: Commandline',
'Intended Audience :: Developers',
'Intended Audience :: System Administrators'
],
)
Raw
TODO
- code activemq.py
- (this will take some time, as there is a lot trial and error involved)
- local for as long as possible (just output on the shell what you're about to execute)
- code instances.py:
- (this will take some time, as there is a lot trial and error involved)
- local for as long as possible (just output on the shell what you're about to execute)
- print the damn thing to YAML (low) (google: convert object tree pyyaml)
- code stress.py:
- (we should be able to build on the experience from before)
- maybe put everything into one outbuild
- should be pretty fast
- install maven on the target machine and have it build the sources there (so we don't have to ship the artifact, should be pretty fast)
- create updated AMI
- with the new toolset
- with the source checked out
- with hyperic working (running) and pretty much configured
- patch ActiveMQ example utils
- patch the source to a way to provide statistics about the progress (logging)
- outbuild calling maven if possible (not important)
- configure Hyperic HQ
- so it actually works with the tools we have
- configure and discover the clients
- find out how to link from buildout directly into the path
- run the tests
--- OPTIONAL --
- Start the clients with boto from localhost
- (this would allow for correct instantiation everytime)
Raw
TopicListener.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* Use in conjunction with TopicPublisher to test the performance of ActiveMQ
* Topics.
*/
public class TopicListener implements MessageListener {
private Connection connection;
private MessageProducer producer;
private Session session;
private int count;
private long start;
private Topic topic;
private Topic control;
private String url = "tcp://localhost:61616";
public static void main(String[] argv) throws Exception {
TopicListener l = new TopicListener();
String[] unknown = CommandLineSupport.setOptions(l, argv);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
l.run();
}
public void run() throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
connection = factory.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
topic = session.createTopic("topictest.messages");
control = session.createTopic("topictest.control");
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(this);
connection.start();
producer = session.createProducer(control);
System.out.println("Waiting for messages...");
}
private static boolean checkText(Message m, String s) {
try {
return m instanceof TextMessage && ((TextMessage)m).getText().equals(s);
} catch (JMSException e) {
e.printStackTrace(System.out);
return false;
}
}
public void onMessage(Message message) {
if (checkText(message, "SHUTDOWN")) {
try {
connection.close();
} catch (Exception e) {
e.printStackTrace(System.out);
}
} else if (checkText(message, "REPORT")) {
// send a report:
try {
long time = System.currentTimeMillis() - start;
String msg = "Received " + count + " in " + time + "ms";
producer.send(session.createTextMessage(msg));
} catch (Exception e) {
e.printStackTrace(System.out);
}
count = 0;
} else {
if (count == 0) {
start = System.currentTimeMillis();
}
if (++count % 1000 == 0) {
System.out.println("Received " + count + " messages.");
}
}
}
public void setUrl(String url) {
this.url = url;
}
}
Raw
TopicPublisher.java
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
import java.util.Arrays;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* Use in conjunction with TopicListener to test the performance of ActiveMQ
* Topics.
*/
public class TopicPublisher implements MessageListener {
private static final char[] DATA = "abcdefghijklmnopqrstuvwxyz".toCharArray();
private final Object mutex = new Object();
private Connection connection;
private Session session;
private MessageProducer publisher;
private Topic topic;
private Topic control;
private String url = "tcp://localhost:61616";
private int size = 256;
private int subscribers = 1;
private int remaining;
private int messages = 10000;
private long delay;
private int batch = 2000;
private byte[] payload;
public static void main(String[] argv) throws Exception {
TopicPublisher p = new TopicPublisher();
String[] unknown = CommandLineSupport.setOptions(p, argv);
if (unknown.length > 0) {
System.out.println("Unknown options: " + Arrays.toString(unknown));
System.exit(-1);
}
p.run();
}
private void run() throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
connection = factory.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
topic = session.createTopic("topictest.messages");
control = session.createTopic("topictest.control");
publisher = session.createProducer(topic);
publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
payload = new byte[size];
for (int i = 0; i < size; i++) {
payload[i] = (byte)DATA[i % DATA.length];
}
session.createConsumer(control).setMessageListener(this);
connection.start();
long[] times = new long[batch];
for (int i = 0; i < batch; i++) {
if (i > 0) {
Thread.sleep(delay * 1000);
}
times[i] = batch(messages);
System.out.println("Batch " + (i + 1) + " of " + batch + " completed in " + times[i] + " ms.");
}
long min = min(times);
long max = max(times);
System.out.println("min: " + min + ", max: " + max + " avg: " + avg(times, min, max));
// request shutdown
publisher.send(session.createTextMessage("SHUTDOWN"));
connection.stop();
connection.close();
}
private long batch(int msgCount) throws Exception {
long start = System.currentTimeMillis();
remaining = subscribers;
publish();
waitForCompletion();
return System.currentTimeMillis() - start;
}
private void publish() throws Exception {
// send events
BytesMessage msg = session.createBytesMessage();
msg.writeBytes(payload);
for (int i = 0; i < messages; i++) {
publisher.send(msg);
if ((i + 1) % 1000 == 0) {
System.out.println("Sent " + (i + 1) + " messages");
}
}
// request report
publisher.send(session.createTextMessage("REPORT"));
}
private void waitForCompletion() throws Exception {
System.out.println("Waiting for completion...");
synchronized (mutex) {
while (remaining > 0) {
mutex.wait();
}
}
}
public void onMessage(Message message) {
synchronized (mutex) {
System.out.println("Received report " + getReport(message) + " " + --remaining + " remaining");
if (remaining == 0) {
mutex.notify();
}
}
}
Object getReport(Message m) {
try {
return ((TextMessage)m).getText();
} catch (JMSException e) {
e.printStackTrace(System.out);
return e.toString();
}
}
static long min(long[] times) {
long min = times.length > 0 ? times[0] : 0;
for (int i = 0; i < times.length; i++) {
min = Math.min(min, times[i]);
}
return min;
}
static long max(long[] times) {
long max = times.length > 0 ? times[0] : 0;
for (int i = 0; i < times.length; i++) {
max = Math.max(max, times[i]);
}
return max;
}
static long avg(long[] times, long min, long max) {
long sum = 0;
for (int i = 0; i < times.length; i++) {
sum += times[i];
}
sum -= min;
sum -= max;
return sum / times.length - 2;
}
public void setBatch(int batch) {
this.batch = batch;
}
public void setDelay(long delay) {
this.delay = delay;
}
public void setMessages(int messages) {
this.messages = messages;
}
public void setSize(int size) {
this.size = size;
}
public void setSubscribers(int subscribers) {
this.subscribers = subscribers;
}
public void setUrl(String url) {
this.url = url;
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment