1 /*

     2  * Copyright 2007 Sun Microsystems, Inc.  All Rights Reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.

     8  *

     9  * This code is distributed in the hope that it will be useful, but WITHOUT

    10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    12  * version 2 for more details (a copy is included in the LICENSE file that

    13  * accompanied this code).

    14  *

    15  * You should have received a copy of the GNU General Public License version

    16  * 2 along with this work; if not, write to the Free Software Foundation,

    17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    18  *

    19  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    20  * CA 95054 USA or visit www.sun.com if you need additional information or

    21  * have any questions.

    22  */

    23 

    24 /*

    25  * @test CustomForwarderTest

    26  * @bug 5108776

    27  * @summary Test that a custom EventForwarder can be added

    28  * @author Eamonn McManus

    29  */

    30 

    31 import java.io.ByteArrayInputStream;

    32 import java.io.ByteArrayOutputStream;

    33 import java.io.IOException;

    34 import java.io.ObjectInputStream;

    35 import java.io.ObjectOutputStream;

    36 import java.lang.management.ManagementFactory;

    37 import java.net.DatagramPacket;

    38 import java.net.DatagramSocket;

    39 import java.net.SocketAddress;

    40 import java.util.Map;

    41 import java.util.concurrent.ArrayBlockingQueue;

    42 import java.util.concurrent.BlockingQueue;

    43 import java.util.concurrent.Semaphore;

    44 import java.util.concurrent.TimeUnit;

    45 import java.util.concurrent.atomic.AtomicBoolean;

    46 import java.util.concurrent.atomic.AtomicInteger;

    47 import java.util.concurrent.atomic.AtomicLong;

    48 import javax.management.MBeanServer;

    49 import javax.management.MBeanServerInvocationHandler;

    50 import javax.management.Notification;

    51 import javax.management.NotificationBroadcasterSupport;

    52 import javax.management.NotificationFilter;

    53 import javax.management.NotificationListener;

    54 import javax.management.ObjectName;

    55 import javax.management.event.EventClient;

    56 import javax.management.event.EventClientDelegate;

    57 import javax.management.event.EventClientDelegateMBean;

    58 import javax.management.event.EventForwarder;

    59 import javax.management.event.EventReceiver;

    60 import javax.management.event.EventRelay;

    61 import javax.management.remote.MBeanServerForwarder;

    62 import javax.management.remote.NotificationResult;

    63 import javax.management.remote.TargetedNotification;

    64 

    65 public class CustomForwarderTest {

    66     public static class UdpEventRelay implements EventRelay {

    67         private final EventClientDelegateMBean delegate;

    68         private final DatagramSocket socket;

    69         private final AtomicBoolean closed = new AtomicBoolean();

    70         private final String clientId;

    71         private EventReceiver receiver;

    72 

    73         public UdpEventRelay(EventClientDelegateMBean delegate)

    74         throws IOException {

    75             this.delegate = delegate;

    76             this.socket = new DatagramSocket();

    77             try {

    78                 clientId = delegate.addClient(

    79                         UdpEventForwarder.class.getName(),

    80                         new Object[] {socket.getLocalSocketAddress()},

    81                         new String[] {SocketAddress.class.getName()});

    82             } catch (IOException e) {

    83                 throw e;

    84             } catch (RuntimeException e) {

    85                 throw e;

    86             } catch (Exception e) {

    87                 final IOException ioe =

    88                         new IOException("Exception creating EventForwarder");

    89                 ioe.initCause(e);

    90                 throw ioe;

    91             }

    92             Thread t = new Thread(new Receiver());

    93             t.setDaemon(true);

    94             t.start();

    95         }

    96 

    97         public String getClientId() throws IOException {

    98             return clientId;

    99         }

   100 

   101         public void setEventReceiver(EventReceiver eventReceiver) {

   102             this.receiver = eventReceiver;

   103         }

   104 

   105         public void stop() throws IOException {

   106             closed.set(true);

   107             socket.close();

   108         }

   109 

   110         private class Receiver implements Runnable {

   111             public void run() {

   112                 byte[] buf = new byte[1024];

   113                 DatagramPacket packet = new DatagramPacket(buf, buf.length);

   114                 while (true) {

   115                     try {

   116                         socket.receive(packet);

   117                     } catch (IOException e) {

   118                         if (closed.get()) {

   119                             System.out.println("Receiver got exception: " + e);

   120                             System.out.println("Normal because it has been closed");

   121                             return;

   122                         } else {

   123                             System.err.println("UNEXPECTED EXCEPTION IN RECEIVER:");

   124                             e.printStackTrace();

   125                             System.exit(1);

   126                         }

   127                     }

   128                     try {

   129                         ByteArrayInputStream bin = new ByteArrayInputStream(buf);

   130                         ObjectInputStream oin = new ObjectInputStream(bin);

   131                         NotificationResult nr = (NotificationResult)

   132                                 oin.readObject();

   133                         receiver.receive(nr);

   134                     } catch (Exception e) {

   135                         System.err.println("UNEXPECTED EXCEPTION IN RECEIVER:");

   136                         e.printStackTrace();

   137                         System.exit(1);

   138                     }

   139                 }

   140             }

   141         }

   142     }

   143 

   144     public static class UdpEventForwarder implements EventForwarder {

   145         private final DatagramSocket socket;

   146         private final AtomicLong seqNo = new AtomicLong(0);

   147         private static volatile boolean drop;

   148 

   149         public UdpEventForwarder(SocketAddress addr) throws IOException {

   150             this.socket = new DatagramSocket();

   151             socket.connect(addr);

   152         }

   153 

   154         public static void setDrop(boolean drop) {

   155             UdpEventForwarder.drop = drop;

   156         }

   157 

   158         public void forward(Notification n, Integer listenerId) throws IOException {

   159             long nextSeqNo = seqNo.incrementAndGet();

   160             long thisSeqNo = nextSeqNo - 1;

   161             TargetedNotification tn = new TargetedNotification(n, listenerId);

   162             NotificationResult nr = new NotificationResult(

   163                     thisSeqNo, nextSeqNo, new TargetedNotification[] {tn});

   164             ByteArrayOutputStream bout = new ByteArrayOutputStream();

   165             ObjectOutputStream oout = new ObjectOutputStream(bout);

   166             oout.writeObject(nr);

   167             oout.close();

   168             byte[] bytes = bout.toByteArray();

   169             DatagramPacket packet = new DatagramPacket(bytes, bytes.length);

   170             if (!drop)

   171                 socket.send(packet);

   172         }

   173 

   174         public void close() throws IOException {

   175             socket.close();

   176         }

   177 

   178         public void setClientId(String clientId) throws IOException {

   179             // Nothing to do.

   180         }

   181     }

   182 

   183     public static interface EmptyMBean {}

   184 

   185     public static class Empty

   186             extends NotificationBroadcasterSupport implements EmptyMBean {

   187         public void send(Notification n) {

   188             super.sendNotification(n);

   189         }

   190     }

   191 

   192     public static void main(String[] args) throws Exception {

   193         MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();

   194         MBeanServerForwarder mbsf = EventClientDelegate.newForwarder();

   195         mbsf.setMBeanServer(mbs);

   196         mbs = mbsf;

   197 

   198         // for 1.5

   199         if (System.getProperty("java.version").startsWith("1.5") &&

   200                 !mbs.isRegistered(EventClientDelegateMBean.OBJECT_NAME)) {

   201             System.out.print("Working on "+System.getProperty("java.version")+

   202                     " register "+EventClientDelegateMBean.OBJECT_NAME);

   203 

   204             mbs.registerMBean(EventClientDelegate.

   205                     getEventClientDelegate(mbs),

   206                     EventClientDelegateMBean.OBJECT_NAME);

   207         }

   208 

   209         ObjectName name = new ObjectName("a:b=c");

   210         Empty mbean = new Empty();

   211         mbs.registerMBean(mbean, name);

   212 

   213         EventClientDelegateMBean delegate = (EventClientDelegateMBean)

   214             MBeanServerInvocationHandler.newProxyInstance(

   215                 mbs,

   216                 EventClientDelegateMBean.OBJECT_NAME,

   217                 EventClientDelegateMBean.class,

   218                 false);

   219         EventRelay relay = new UdpEventRelay(delegate);

   220         EventClient client = new EventClient(delegate, relay, null, null, 0L);

   221 

   222         final Semaphore lostCountSema = new Semaphore(0);

   223         NotificationListener lostListener = new NotificationListener() {

   224             public void handleNotification(Notification notification, Object handback) {

   225                 if (notification.getType().equals(EventClient.NOTIFS_LOST)) {

   226                     System.out.println("Got lost-notifs notif: count=" +

   227                             notification.getUserData());

   228                     lostCountSema.release(((Long) notification.getUserData()).intValue());

   229                 } else

   230                     System.out.println("Mysterious EventClient notif: " + notification);

   231             }

   232         };

   233         client.addEventClientListener(lostListener, null, null);

   234 

   235         final BlockingQueue<Notification> notifQueue =

   236                 new ArrayBlockingQueue<Notification>(10);

   237         NotificationListener countListener = new NotificationListener() {

   238             public void handleNotification(Notification notification, Object handback) {

   239                 System.out.println("Received: " + notification);

   240                 notifQueue.add(notification);

   241                 if (!"tiddly".equals(handback)) {

   242                     System.err.println("TEST FAILED: bad handback: " + handback);

   243                     System.exit(1);

   244                 }

   245             }

   246         };

   247 

   248         final AtomicInteger filterCount = new AtomicInteger(0);

   249         NotificationFilter countFilter = new NotificationFilter() {

   250             private static final long serialVersionUID = 1234L;

   251 

   252             public boolean isNotificationEnabled(Notification notification) {

   253                 System.out.println("Filter called for: " + notification);

   254                 filterCount.incrementAndGet();

   255                 return true;

   256             }

   257         };

   258 

   259         client.addNotificationListener(name, countListener, countFilter, "tiddly");

   260 

   261         assertEquals("Initial notif count", 0, notifQueue.size());

   262         assertEquals("Initial filter count", 0, filterCount.get());

   263 

   264         Notification n = nextNotif(name);

   265         mbean.send(n);

   266 

   267         System.out.println("Waiting for notification to arrive...");

   268 

   269         Notification n1 = notifQueue.poll(10, TimeUnit.SECONDS);

   270 

   271         assertEquals("Received notif", n, n1);

   272         assertEquals("Notif queue size after receive", 0, notifQueue.size());

   273         assertEquals("Filter count after notif", 1, filterCount.get());

   274         assertEquals("Lost notif count", 0, lostCountSema.availablePermits());

   275 

   276         System.out.println("Dropping notifs");

   277 

   278         UdpEventForwarder.setDrop(true);

   279         for (int i = 0; i < 3; i++)

   280             mbean.send(nextNotif(name));

   281         UdpEventForwarder.setDrop(false);

   282 

   283         Thread.sleep(2);

   284         assertEquals("Notif queue size after drops", 0, notifQueue.size());

   285 

   286         System.out.println("Turning off dropping and sending a notif");

   287         n = nextNotif(name);

   288         mbean.send(n);

   289 

   290         System.out.println("Waiting for dropped notifications to be detected...");

   291         boolean acquired = lostCountSema.tryAcquire(3, 5, TimeUnit.SECONDS);

   292         assertEquals("Correct count of lost notifs", true, acquired);

   293 

   294         n1 = notifQueue.poll(10, TimeUnit.SECONDS);

   295         assertEquals("Received non-dropped notif", n, n1);

   296 

   297         assertEquals("Notif queue size", 0, notifQueue.size());

   298         assertEquals("Filter count after drops", 5, filterCount.get());

   299 

   300         Thread.sleep(10);

   301         assertEquals("Further lost-notifs", 0, lostCountSema.availablePermits());

   302 

   303         client.close();

   304 

   305         System.out.println("TEST PASSED");

   306     }

   307 

   308     private static AtomicLong nextSeqNo = new AtomicLong(0);

   309     private static Notification nextNotif(ObjectName name) {

   310         long n = nextSeqNo.incrementAndGet();

   311         return new Notification("type", name, n, "" + n);

   312     }

   313 

   314     private static void assertEquals(String what, Object expected, Object got) {

   315         if (equals(expected, got))

   316             System.out.println(what + " = " + expected + ", as expected");

   317         else {

   318             Map<Thread, StackTraceElement[]> traces = Thread.getAllStackTraces();

   319             for (Thread t : traces.keySet()) {

   320                 System.out.println(t.getName());

   321                 for (StackTraceElement elmt : traces.get(t)) {

   322                     System.out.println("    " + elmt);

   323                 }

   324             }

   325             throw new RuntimeException(

   326                     "TEST FAILED: " + what + " is " + got + "; should be " +

   327                     expected);

   328         }

   329     }

   330 

   331     private static boolean equals(Object expected, Object got) {

   332         if (!(expected instanceof Notification))

   333             return expected.equals(got);

   334         if (expected.getClass() != got.getClass())

   335             return false;

   336         // Notification doesn't override Object.equals so two distinct

   337         // notifs are never equal even if they have the same contents.

   338         // Although the test doesn't serialize the notifs, if at some

   339         // stage it did then it would fail because the deserialized notif

   340         // was not equal to the original one.  Therefore we compare enough

   341         // notif fields to detect when notifs really are different.

   342         Notification en = (Notification) expected;

   343         Notification gn = (Notification) got;

   344         return (en.getType().equals(gn.getType()) &&

   345                 en.getSource().equals(gn.getSource()) &&

   346                 en.getSequenceNumber() == gn.getSequenceNumber());

   347     }

   348 }