1 /*

     2  * Copyright 2007-2008 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 6759619

    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.MBeanNotificationInfo;

    49 import javax.management.MBeanServer;

    50 import javax.management.MBeanServerInvocationHandler;

    51 import javax.management.Notification;

    52 import javax.management.NotificationBroadcasterSupport;

    53 import javax.management.NotificationFilter;

    54 import javax.management.NotificationListener;

    55 import javax.management.ObjectName;

    56 import javax.management.event.EventClient;

    57 import javax.management.event.EventClientDelegate;

    58 import javax.management.event.EventClientDelegateMBean;

    59 import javax.management.event.EventForwarder;

    60 import javax.management.event.EventReceiver;

    61 import javax.management.event.EventRelay;

    62 import javax.management.remote.MBeanServerForwarder;

    63 import javax.management.remote.NotificationResult;

    64 import javax.management.remote.TargetedNotification;

    65 

    66 public class CustomForwarderTest {

    67     public static class UdpEventRelay implements EventRelay {

    68         private final EventClientDelegateMBean delegate;

    69         private final DatagramSocket socket;

    70         private final AtomicBoolean closed = new AtomicBoolean();

    71         private final String clientId;

    72         private EventReceiver receiver;

    73 

    74         public UdpEventRelay(EventClientDelegateMBean delegate)

    75         throws IOException {

    76             this.delegate = delegate;

    77             this.socket = new DatagramSocket();

    78             try {

    79                 clientId = delegate.addClient(

    80                         UdpEventForwarder.class.getName(),

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

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

    83             } catch (IOException e) {

    84                 throw e;

    85             } catch (RuntimeException e) {

    86                 throw e;

    87             } catch (Exception e) {

    88                 final IOException ioe =

    89                         new IOException("Exception creating EventForwarder");

    90                 ioe.initCause(e);

    91                 throw ioe;

    92             }

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

    94             t.setDaemon(true);

    95             t.start();

    96         }

    97 

    98         public String getClientId() throws IOException {

    99             return clientId;

   100         }

   101 

   102         public void setEventReceiver(EventReceiver eventReceiver) {

   103             this.receiver = eventReceiver;

   104         }

   105 

   106         public void stop() throws IOException {

   107             closed.set(true);

   108             socket.close();

   109         }

   110 

   111         void simulateNonFatal() {

   112             receiver.nonFatal(new Exception("NonFatal"));

   113         }

   114 

   115         void simulateFailed() {

   116             receiver.failed(new Error("Failed"));

   117         }

   118 

   119         private class Receiver implements Runnable {

   120             public void run() {

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

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

   123                 while (true) {

   124                     try {

   125                         socket.receive(packet);

   126                     } catch (IOException e) {

   127                         if (closed.get()) {

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

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

   130                             return;

   131                         } else {

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

   133                             e.printStackTrace();

   134                             System.exit(1);

   135                         }

   136                     }

   137                     try {

   138                         ByteArrayInputStream bin = new ByteArrayInputStream(buf);

   139                         ObjectInputStream oin = new ObjectInputStream(bin);

   140                         NotificationResult nr = (NotificationResult)

   141                                 oin.readObject();

   142                         receiver.receive(nr);

   143                     } catch (Exception e) {

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

   145                         e.printStackTrace();

   146                         System.exit(1);

   147                     }

   148                 }

   149             }

   150         }

   151     }

   152 

   153     public static class UdpEventForwarder implements EventForwarder {

   154         private final DatagramSocket socket;

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

   156         private static volatile boolean drop;

   157 

   158         public UdpEventForwarder(SocketAddress addr) throws IOException {

   159             this.socket = new DatagramSocket();

   160             socket.connect(addr);

   161         }

   162 

   163         public static void setDrop(boolean drop) {

   164             UdpEventForwarder.drop = drop;

   165         }

   166 

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

   168             long nextSeqNo = seqNo.incrementAndGet();

   169             long thisSeqNo = nextSeqNo - 1;

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

   171             NotificationResult nr = new NotificationResult(

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

   173             ByteArrayOutputStream bout = new ByteArrayOutputStream();

   174             ObjectOutputStream oout = new ObjectOutputStream(bout);

   175             oout.writeObject(nr);

   176             oout.close();

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

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

   179             if (!drop)

   180                 socket.send(packet);

   181         }

   182 

   183         public void close() throws IOException {

   184             socket.close();

   185         }

   186 

   187         public void setClientId(String clientId) throws IOException {

   188             // Nothing to do.

   189         }

   190     }

   191 

   192     public static interface EmptyMBean {}

   193 

   194     public static class Empty

   195             extends NotificationBroadcasterSupport implements EmptyMBean {

   196         public void send(Notification n) {

   197             super.sendNotification(n);

   198         }

   199     }

   200 

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

   202         MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();

   203         MBeanServerForwarder mbsf = EventClientDelegate.newForwarder(mbs, null);

   204         mbs = mbsf;

   205 

   206         // for 1.5

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

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

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

   210                     " register "+EventClientDelegateMBean.OBJECT_NAME);

   211 

   212             mbs.registerMBean(EventClientDelegate.

   213                     getEventClientDelegate(mbs),

   214                     EventClientDelegateMBean.OBJECT_NAME);

   215         }

   216 

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

   218         Empty mbean = new Empty();

   219         mbs.registerMBean(mbean, name);

   220 

   221         EventClientDelegateMBean delegate = (EventClientDelegateMBean)

   222             MBeanServerInvocationHandler.newProxyInstance(

   223                 mbs,

   224                 EventClientDelegateMBean.OBJECT_NAME,

   225                 EventClientDelegateMBean.class,

   226                 false);

   227         UdpEventRelay relay = new UdpEventRelay(delegate);

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

   229 

   230         final Semaphore lostCountSema = new Semaphore(0);

   231         final BlockingQueue<Notification> nonFatalNotifs =

   232                 new ArrayBlockingQueue<Notification>(1);

   233         final BlockingQueue<Notification> failedNotifs =

   234                 new ArrayBlockingQueue<Notification>(1);

   235         NotificationListener lostListener = new NotificationListener() {

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

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

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

   239                             notification.getUserData());

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

   241                 } else if (notification.getType().equals(EventClient.NONFATAL)) {

   242                     System.out.println("Got nonFatal notif");

   243                     nonFatalNotifs.add(notification);

   244                 } else if (notification.getType().equals(EventClient.FAILED)) {

   245                     System.out.println("Got failed notif");

   246                     failedNotifs.add(notification);

   247                 } else

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

   249             }

   250         };

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

   252 

   253         final BlockingQueue<Notification> notifQueue =

   254                 new ArrayBlockingQueue<Notification>(10);

   255         NotificationListener countListener = new NotificationListener() {

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

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

   258                 notifQueue.add(notification);

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

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

   261                     System.exit(1);

   262                 }

   263             }

   264         };

   265 

   266         final AtomicInteger filterCount = new AtomicInteger(0);

   267         NotificationFilter countFilter = new NotificationFilter() {

   268             private static final long serialVersionUID = 1234L;

   269 

   270             public boolean isNotificationEnabled(Notification notification) {

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

   272                 filterCount.incrementAndGet();

   273                 return true;

   274             }

   275         };

   276 

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

   278 

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

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

   281 

   282         Notification n = nextNotif(name);

   283         mbean.send(n);

   284 

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

   286 

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

   288 

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

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

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

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

   293 

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

   295 

   296         UdpEventForwarder.setDrop(true);

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

   298             mbean.send(nextNotif(name));

   299         UdpEventForwarder.setDrop(false);

   300 

   301         Thread.sleep(2);

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

   303 

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

   305         n = nextNotif(name);

   306         mbean.send(n);

   307 

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

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

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

   311 

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

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

   314 

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

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

   317 

   318         Thread.sleep(10);

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

   320 

   321         System.out.println("Testing error notifs");

   322         relay.simulateNonFatal();

   323         n = nonFatalNotifs.poll(10, TimeUnit.SECONDS);

   324         assertEquals("Exception message for non-fatal exception", "NonFatal",

   325                 ((Throwable) n.getSource()).getMessage());

   326         relay.simulateFailed();

   327         n = failedNotifs.poll(10, TimeUnit.SECONDS);

   328         assertEquals("Exception message for failed exception", "Failed",

   329                 ((Throwable) n.getSource()).getMessage());

   330 

   331         // 6759619

   332         System.out.println("Test EventClient.getEventClientNotificationInfo");

   333         MBeanNotificationInfo[] mbnis = client.getEventClientNotificationInfo();

   334         final String[] expectedTypes = {

   335             EventClient.NOTIFS_LOST, EventClient.NONFATAL, EventClient.FAILED

   336         };

   337     check:

   338         for (String type : expectedTypes) {

   339             for (MBeanNotificationInfo mbni : mbnis) {

   340                 for (String t : mbni.getNotifTypes()) {

   341                     if (type.equals(t)) {

   342                         System.out.println("...found " + type);

   343                         continue check;

   344                     }

   345                 }

   346             }

   347             throw new Exception("TEST FAILED: Did not find notif type " + type);

   348         }

   349 

   350         client.close();

   351 

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

   353     }

   354 

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

   356     private static Notification nextNotif(ObjectName name) {

   357         long n = nextSeqNo.incrementAndGet();

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

   359     }

   360 

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

   362         if (equals(expected, got))

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

   364         else {

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

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

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

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

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

   370                 }

   371             }

   372             throw new RuntimeException(

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

   374                     expected);

   375         }

   376     }

   377 

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

   379         if (!(expected instanceof Notification))

   380             return expected.equals(got);

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

   382             return false;

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

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

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

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

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

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

   389         Notification en = (Notification) expected;

   390         Notification gn = (Notification) got;

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

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

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

   394     }

   395 }